import { EventEmitter } from 'events' ; import logdown from 'logdown' ; import type { CloseEvent , ErrorEvent , Event } from 'reconnecting-websocket' ; import { HttpClient , NetworkError } from '../http/' ; import { InvalidTokenError , MissingCookieError } from '../auth/' ; import { ReconnectingWebsocket , WEBSOCKET_STATE } from './ReconnectingWebsocket' ; import type { Notification } from '../notification/' ; enum TOPIC { ON_ERROR = 'WebSocketClient.TOPIC.ON_ERROR' , ON_INVALID_TOKEN = 'WebSocketClient.TOPIC.ON_INVALID_TOKEN' , ON_MESSAGE = 'WebSocketClient.TOPIC.ON_MESSAGE' , ON_STATE_CHANGE = 'WebSocketClient.TOPIC.ON_STATE_CHANGE' } export interface WebSocketClient { on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; } export class AbortHandler { private aborted = false ; abort = ( ) => { this . aborted = true ; } ; isAborted = ( ) => this . aborted ; } export type OnConnect = ( abortHandler : AbortHandler ) => Promise < void > ; export class WebSocketClient extends EventEmitter { private clientId ? : string ; private isRefreshingAccessToken : boolean ; private readonly baseUrl : string ; private readonly logger : logdown . Logger ; private readonly socket : ReconnectingWebsocket ; private websocketState : WEBSOCKET_STATE ; public client : HttpClient ; private isSocketLocked : boolean ; private bufferedMessages : string [ ] ; private onConnect : ( ) => Promise < void > = ( ) => Promise . resolve ( ) ; private abortHandler ? : AbortHandler ; public static readonly TOPIC = TOPIC ; constructor ( baseUrl , client ) { super ( ) ; this . bufferedMessages = [ ] ; this . isSocketLocked = false ; this . baseUrl = baseUrl ; this . client = client ; this . isRefreshingAccessToken = false ; this . socket = new ReconnectingWebsocket ( this . onReconnect ) ; this . websocketState = this . socket . getState ( ) ; this . logger = logdown ( '@wireapp/api-client/tcp/WebSocketClient' , { logger : console , markdown : false , } ) ; } private onStateChange ( newState ) { if ( newState !== this . websocketState ) { this . websocketState = newState ; this . emit ( WebSocketClient . TOPIC . ON_STATE_CHANGE , this . websocketState ) ; } } private readonly onMessage = ( data ) => { if ( this . isLocked ( ) ) { this . bufferedMessages . push ( data ) ; } else { const notification = JSON . parse ( data ) ; this . emit ( WebSocketClient . TOPIC . ON_MESSAGE , notification ) ; } } ; private readonly onError = async ( error ) => { this . onStateChange ( this . socket . getState ( ) ) ; this . emit ( WebSocketClient . TOPIC . ON_ERROR , error ) ; await this . refreshAccessToken ( ) ; } ; private readonly onReconnect = async ( ) => { void this . onConnect ( ) ; return this . buildWebSocketUrl ( ) ; } ; private readonly onOpen = ( event ) => { this . onStateChange ( this . socket . getState ( ) ) ; } ; private readonly onClose = ( event ) => { this . abortHandler ?. abort ( ) ; this . bufferedMessages = [ ] ; this . onStateChange ( this . socket . getState ( ) ) ; } ; param clientId param onConnect public async connect ( clientId ? , onConnect ? ) { if ( onConnect ) { this . onConnect = async ( ) => { this . abortHandler = new AbortHandler ( ) ; try { this . logger . info ( 'Calling "onConnect"' ) ; await onConnect ( this . abortHandler ) ; } catch ( error ) { this . logger . warn ( `Error during execution of "onConnect"` , error ) ; this . emit ( WebSocketClient . TOPIC . ON_ERROR , error ) ; } this . onStateChange ( this . socket . getState ( ) ) ; } ; } this . clientId = clientId ; this . socket . setOnMessage ( this . onMessage ) ; this . socket . setOnError ( this . onError ) ; this . socket . setOnOpen ( this . onOpen ) ; this . socket . setOnClose ( this . onClose ) ; this . socket . connect ( ) ; return this ; } private async refreshAccessToken ( ) { if ( this . isRefreshingAccessToken ) { return ; } this . isRefreshingAccessToken = true ; try { await this . client . refreshAccessToken ( ) ; } catch ( error ) { if ( error instanceof NetworkError ) { this . logger . warn ( error ) ; } else if ( error instanceof InvalidTokenError || error instanceof MissingCookieError ) { this . logger . warn ( `[WebSocket] Cannot renew access token because cookie/token is invalid: ${ error . message }` , error ) ; this . emit ( WebSocketClient . TOPIC . ON_INVALID_TOKEN , error ) ; } else { this . emit ( WebSocketClient . TOPIC . ON_ERROR , error ) ; } } finally { this . isRefreshingAccessToken = false ; } } public disconnect ( reason ? , keepClosed = true ) { if ( this . socket ) { this . socket . disconnect ( reason , keepClosed ) ; } } /**
     * Unlocks the websocket.
     * When unlocking the websocket all buffered messages between
     * connecting the websocket and the unlocking the websocket will be emitted.
     */ public readonly unlock = ( ) => { this . logger . info ( `Unlocking WebSocket - Emitting "${ this . bufferedMessages . length }" unprocessed messages` ) ; this . isSocketLocked = false ; for ( const bufferedMessage of this . bufferedMessages ) { this . onMessage ( bufferedMessage ) ; } this . bufferedMessages = [ ] ; } ; /**
     * Locks the websocket so messages are buffered instead of being emitted.
     * Once the websocket gets unlocked buffered messages get emitted.
     * This behaviour is needed in order to not miss any messages
     * during fetching notifications from the notification stream.
     */ public readonly lock = ( ) => { this . logger . info ( 'Locking WebSocket' ) ; this . isSocketLocked = true ; } ; public isLocked ( ) { return this . isSocketLocked ; } private buildWebSocketUrl ( ) { const store = this . client . accessTokenStore . accessToken ; const token = store && store . access_token ? store . access_token : '' ; if ( ! token ) { this . logger . warn ( 'Reconnecting WebSocket with unset token' ) ; } let url = `${ this . baseUrl }/await?access_token=${ token }` ; if ( this . clientId ) { url += `&client=${ this . clientId }` ; } return url ; } }