import { TimeUtil } from '@wireapp/commons' ; import logdown from 'logdown' ; import RWS , { CloseEvent , ErrorEvent , Event , Options } from 'reconnecting-websocket' ; const NodeWebSocket = require ( 'ws' ) ; import * as buffer from '../shims/node/buffer' ; export enum CloseEventCode { NORMAL_CLOSURE = 1000 , GOING_AWAY = 1001 , PROTOCOL_ERROR = 1002 , UNSUPPORTED_DATA = 1003 } export enum WEBSOCKET_STATE { CONNECTING = 0 , OPEN = 1 , CLOSING = 2 , CLOSED = 3 } export enum PingMessage { PING = 'ping' , PONG = 'pong' } export class ReconnectingWebsocket { private static readonly RECONNECTING_OPTIONS : Options = { WebSocket : typeof window !== 'undefined' ? WebSocket : NodeWebSocket , connectionTimeout : TimeUtil . TimeInMillis . SECOND * 4 , debug : false , maxReconnectionDelay : TimeUtil . TimeInMillis . SECOND * 10 , maxRetries : Infinity , minReconnectionDelay : TimeUtil . TimeInMillis . SECOND * 4 , reconnectionDelayGrowFactor : 1.3 , } ; private readonly logger : logdown . Logger ; private socket ? : RWS ; private pingerId ? : NodeJS . Timeout ; private readonly PING_INTERVAL = TimeUtil . TimeInMillis . SECOND * 20 ; private hasUnansweredPing : boolean ; private onOpen ? : ( event : Event ) => void ; private onMessage ? : ( data : string ) => void ; private onError ? : ( error : ErrorEvent ) => void ; private onClose ? : ( event : CloseEvent ) => void ; constructor ( private readonly onReconnect , options = { } ) { this . logger = logdown ( '@wireapp/api-client/tcp/ReconnectingWebsocket' , { logger : console , markdown : false , } ) ; if ( options . pingInterval ) { this . PING_INTERVAL = options . pingInterval ; } this . hasUnansweredPing = false ; } private readonly internalOnError = ( error ) => { this . logger . warn ( 'WebSocket connection error' , error ) ; if ( this . onError ) { this . onError ( error ) ; } } ; private readonly internalOnMessage = ( event ) => { this . logger . debug ( 'Incoming message' ) ; const data = buffer . bufferToString ( event . data ) ; if ( data === PingMessage . PONG ) { this . logger . debug ( 'Received pong from WebSocket' ) ; this . hasUnansweredPing = false ; } else if ( this . onMessage ) { this . onMessage ( data ) ; } } ; private readonly internalOnOpen = ( event ) => { this . logger . debug ( 'WebSocket opened' ) ; if ( this . socket ) { this . socket . binaryType = 'arraybuffer' ; } if ( this . onOpen ) { this . onOpen ( event ) ; } } ; private readonly internalOnReconnect = async ( ) : Promise < string > => { this . logger . debug ( 'Connecting to WebSocket' ) ; this . startPinging ( ) ; return this . onReconnect ( ) ; } ; private readonly internalOnClose = ( event ) => { this . logger . debug ( 'WebSocket closed' ) ; this . stopPinging ( ) ; if ( this . onClose ) { this . onClose ( event ) ; } } ; public connect ( ) { this . socket = this . getReconnectingWebsocket ( ) ; this . socket . onmessage = this . internalOnMessage ; this . socket . onerror = this . internalOnError ; this . socket . onopen = this . internalOnOpen ; this . socket . onclose = this . internalOnClose ; } public send ( message ) { if ( this . socket ) { this . socket . send ( message ) ; } } private startPinging ( ) { this . stopPinging ( ) ; this . hasUnansweredPing = false ; this . pingerId = setInterval ( this . sendPing , this . PING_INTERVAL ) ; } private stopPinging ( ) { if ( this . pingerId ) { clearInterval ( this . pingerId ) ; } } private readonly sendPing = ( ) : void => { if ( this . socket ) { if ( this . hasUnansweredPing ) { this . logger . warn ( 'Ping interval check failed' ) ; this . stopPinging ( ) ; return ; } this . hasUnansweredPing = true ; this . send ( PingMessage . PING ) ; } } ; public getState ( ) { return this . socket ? this . socket . readyState : WEBSOCKET_STATE . CLOSED ; } public disconnect ( reason = 'Closed by client' , keepClosed = true ) { if ( this . socket ) { this . logger . info ( `Disconnecting from WebSocket (reason: "${ reason }")` ) ; ( this . socket as any ) . close ( CloseEventCode . NORMAL_CLOSURE , reason , { delay : 0 , fastClose : true , keepClosed , } ) ; } } private getReconnectingWebsocket ( ) { return new RWS ( this . internalOnReconnect , undefined , ReconnectingWebsocket . RECONNECTING_OPTIONS ) ; } public setOnOpen ( onOpen ) { this . onOpen = onOpen ; } public setOnMessage ( onMessage ) { this . onMessage = onMessage ; } public setOnError ( onError ) { this . onError = onError ; } public setOnClose ( onClose ) { this . onClose = onClose ; } }