import PakoTypes from '@types/pako' ; import WebSocket from 'ws' ; import { TypedEmitter } from 'tiny-typed-emitter' ; import { WebSocketClientDestroyOptions } from '@src/gateway/interfaces/WebSocketClientDestroyOptions' ; import { WebSocketClientEventsHandlers } from '@src/gateway/interfaces/WebSocketClientEventsHandlers' ; import { CompletedGatewayOptions } from '@src/gateway/interfaces/CompletedGatewayOptions' ; import { WebSocketPacket } from '@src/gateway/interfaces/WebSocketPacket' ; import { GatewayOpCodes , GatewaySendPayloadLike } from '@discordoo/providers' ; import { WebSocketClientEvents , WebSocketClientStates , WebSocketStates } from '@src/constants' ; import { WebSocketManager } from '@src/gateway/WebSocketManager' ; import { WebSocketUtils } from '@src/utils/WebSocketUtils' ; import { DiscordooError } from '@src/utils/DiscordooError' ; import { identify } from '@src/gateway/client/identify' ; import { message } from '@src/gateway/client/message' ; import { packet } from '@src/gateway/client/packet' ; import { error } from '@src/gateway/client/error' ; import { close } from '@src/gateway/client/close' ; import { open } from '@src/gateway/client/open' ; export class WebSocketClient extends TypedEmitter < WebSocketClientEventsHandlers > { private socket ? : WebSocket ; private readonly options : CompletedGatewayOptions ; private inflate ? : PakoTypes . Inflate ; private _heartbeatInterval ? : ReturnType < typeof setInterval > ; private _handshakeTimeout ? : ReturnType < typeof setTimeout > ; public manager : WebSocketManager ; public status : WebSocketClientStates ; public id : number ; public sessionId ? : string ; public sequence = - 1 ; public closeSequence = - 1 ; public expectedGuilds : Set < any > = new Set ( ) ; public missedHeartbeats = 0 ; public ping = - 1 ; public lastPingTimestamp = Date . now ( ) ; constructor ( manager , id ) { super ( ) ; this . options = manager . options ; this . manager = manager ; this . status = WebSocketClientStates . CREATED ; this . id = id ; } public connect ( ) { return new Promise < void > ( ( resolve , reject ) => { if ( ! this . options ?. url ) return reject ( ) ; if ( this . socket ?. readyState === WebSocketStates . OPEN && this . status === WebSocketClientStates . READY ) { resolve ( ) ; } if ( this . options . compress ) { if ( ! WebSocketUtils . pako ) { throw new DiscordooError ( 'WebSocketShard ' + this . id , 'gateway compression requires pako/zlib-sync module installed. npm i pako@1.0.11 / npm i zlib-sync' ) ; } else { this . inflate = new WebSocketUtils . pako . Inflate ( { chunkSize : 128 * 1024 } ) ; } } if ( this . options . encoding === 'etf' && WebSocketUtils . encoding !== 'etf' ) { throw new DiscordooError ( 'WebSocketShard ' + this . id , 'cannot use etf encoding without erlpack installed.' ) ; } const cleanup = ( ) => { cleanupOrListen ( this ) ; } ; const ready = ( ) => { cleanup ( ) ; resolve ( ) ; } , resumed = ( ) => { cleanup ( ) ; resolve ( ) ; } , invalid = ( ) => { cleanup ( ) ; reject ( ) ; } , closed = ( ) => { cleanup ( ) ; reject ( ) ; } ; /**
             * when we call WebSocketClient#connect, it returns a promise.
             * this function resolves promise when shard successfully connected,
             * and rejects promise when shard failed to connect for some reason.
             * when promise resolved/rejected, it removes the used before listeners.
             * */ function cleanupOrListen ( client , listen = false ) { const direction = listen ? client . once : client . removeListener ; direction . call ( client , WebSocketClientEvents . WS_CLOSED , closed ) ; direction . call ( client , WebSocketClientEvents . READY , ready ) ; direction . call ( client , WebSocketClientEvents . RESUMED , resumed ) ; direction . call ( client , WebSocketClientEvents . INVALID_SESSION , invalid ) ; direction . call ( client , WebSocketClientEvents . DESTROYED , invalid ) ; } cleanupOrListen ( this , true ) ; if ( this . socket ?. readyState === WebSocketStates . OPEN ) { return this . identify ( ) ; } else if ( this . socket ) { this . socket = undefined ; } this . status = this . status === WebSocketClientStates . DISCONNECTED ? WebSocketClientStates . RECONNECTING : WebSocketClientStates . CONNECTING ; try { this . socket = new WebSocket ( this . options . url ) ; this . handshakeTimeout ( ) ; this . socket . onopen = this . onOpen . bind ( this ) ; this . socket . onclose = this . onClose . bind ( this ) ; this . socket . onerror = this . onError . bind ( this ) ; this . socket . onmessage = this . onMessage . bind ( this ) ; } catch ( e ) { this . status = WebSocketClientStates . DISCONNECTED ; reject ( ) ; } } ) ; } identify ( ) { const data = identify ( this , this . options ) ; this . socketSend ( data ) ; } /**
     * this method operates with handshake timeout:
     * when handshake did not occur at the specified time,
     * it reconnects the client
     * */ public handshakeTimeout ( create = true ) { if ( this . _handshakeTimeout ) { clearTimeout ( this . _handshakeTimeout ) ; } else if ( create ) { this . _handshakeTimeout = setTimeout ( ( ) => { this . destroy ( { reconnect : true } ) ; } , 15000 ) ; } } /** this method operates with heartbeat interval: it can delete interval, or create new one */ public heartbeatInterval ( ms ) { if ( ms < 0 ) { if ( this . _heartbeatInterval ) clearInterval ( this . _heartbeatInterval ) ; this . _heartbeatInterval = undefined ; } else { this . _heartbeatInterval = setInterval ( ( ) => { this . heartbeat ( ) ; } , ms ) ; } } public heartbeat ( shouldIgnoreAck = [ WebSocketClientStates . RESUMING , WebSocketClientStates . IDENTIFYING , WebSocketClientStates . WAITING_FOR_GUILDS ] . includes ( this . status ) ) { if ( this . missedHeartbeats > 1 ) { this . destroy ( { reconnect : true } ) ; } if ( this . status === WebSocketClientStates . RESUMING ) return ; if ( ! shouldIgnoreAck ) this . missedHeartbeats += 1 ; this . lastPingTimestamp = Date . now ( ) ; this . socketSend ( { op : GatewayOpCodes . HEARTBEAT , d : this . sequence } ) ; } public socketSend ( data ) { if ( ! this . socket ) { this . emit ( WebSocketClientEvents . WS_SEND_ERROR , new Error ( 'Tried to send packet, but no WebSocket was found' ) , data ) ; return ; } this . socket . send ( WebSocketUtils . pack ( data ) , err => { if ( err ) this . emit ( WebSocketClientEvents . WS_SEND_ERROR , err , data ) ; } ) ; } private onMessage ( event ) { const data = message ( this , event , this . options , this . inflate ) ; if ( data ) this . onPacket ( data ) ; } private onPacket ( payload ) { packet ( this , payload ) ; } private onClose ( event ) { close ( this , event ) ; } private onOpen ( event ) { open ( this , event ) ; } private onError ( event ) { error ( this , event ) ; } public destroy ( options = { } ) { if ( this . socket && this . socket . readyState !== WebSocketStates . CLOSED && this . socket . readyState !== WebSocketStates . CLOSING ) { try { if ( options . reconnect && this . sessionId ) { if ( this . options . useReconnectOnly ) { this . sessionId = undefined ; this . sequence = - 1 ; this . socket . close ( 1000 , 'ddoo: reconnect without resume' ) ; } else { this . closeSequence = this . sequence ; this . sequence = - 1 ; this . socket . close ( 4901 , 'ddoo: reconnect with resume' ) ; } } } catch ( e ) { this . emit ( WebSocketClientEvents . WS_CLOSE_ERROR , e ) ; } } else { if ( options . code !== 1000 && this . sequence > 0 ) { this . closeSequence = this . sequence ; this . sequence = - 1 ; } this . socket ?. terminate ( ) ; } this . cleanup ( ) ; if ( options . reconnect ) this . emit ( WebSocketClientEvents . RECONNECT_ME , true ) ; } public get ready ( ) { return this . status === WebSocketClientStates . READY ; } private cleanup ( ) { this . heartbeatInterval ( - 1 ) ; this . handshakeTimeout ( false ) ; this . inflate = undefined ; this . missedHeartbeats = 0 ; if ( ! this . socket ) return ; this . socket . onmessage = this . socket . onclose = this . socket . onopen = this . socket . onerror = ( ) => { } ; this . socket . terminate ( ) ; } }