import Trebuchet , { Data , TrebuchetSettings } from './Trebuchet' ; type Encode = ( msg : any ) => Data ; type Decode = ( msg : any ) => { message : any ; mid ? : number ; } ; export interface WSSettings extends TrebuchetSettings { getUrl : ( ) => string ; encode ? : Encode ; decode ? : Decode ; } const PING = 57 ; const PONG = new Uint8Array ( [ 65 ] ) ; const MAX_MID = 2 ** 31 - 1 ; const MAX_IGNORE_LEN = 1000 ; export const TREBUCHET_WS = 'trebuchet-ws' ; const isPing = ( data ) => { if ( typeof data === 'string' ) return false ; const buffer = new Uint8Array ( data ) ; return buffer . length === 1 && buffer [ 0 ] === PING ; } ; const defaultDecode = ( msg ) => { const parsedData = JSON . parse ( msg ) ; if ( ! Array . isArray ( parsedData ) ) return { message : parsedData } ; const [ message , mid ] = parsedData ; return { message , mid } ; } ; class SocketTrebuchet extends Trebuchet { ws ! : WebSocket ; private readonly getUrl : ( ) => string ; private encode : Encode ; private decode : Decode ; private mqTimer : number | undefined ; constructor ( settings ) { super ( settings ) ; const { decode , encode } = settings ; this . encode = encode || JSON . stringify ; this . decode = decode || defaultDecode ; this . getUrl = settings . getUrl ; this . setup ( ) ; } private keepAlive ( ) { clearTimeout ( this . keepAliveTimeoutId ) ; this . keepAliveTimeoutId = window . setTimeout ( ( ) => { this . keepAliveTimeoutId = undefined ; this . ws . close ( 1000 ) ; } , this . timeout * 1.5 ) ; } private sendAck ( mid ) { const ack = new Uint8Array ( 4 ) ; const view = new DataView ( ack . buffer ) ; const ackId = mid << 1 ; view . setUint32 ( 0 , ackId , true ) ; this . ws . send ( ack ) ; } private sendReq ( mid ) { const req = new Uint8Array ( 4 ) ; const view = new DataView ( req . buffer ) ; const reqId = ( mid << 1 ) | 1 ; view . setUint32 ( 0 , reqId , true ) ; this . ws . send ( req ) ; } private releaseNextRobustMessage ( ) { const nextId = this . lastMid + 1 ; const message = this . robustQueue [ nextId ] ; if ( ! message ) { this . requestedMids . length = 0 ; return ; } delete this . robustQueue [ nextId ] ; this . lastMid = nextId ; this . emit ( 'data' , message ) ; this . releaseNextRobustMessage ( ) ; } protected setup ( ) { this . ws = new WebSocket ( this . getUrl ( ) , TREBUCHET_WS ) ; this . ws . binaryType = 'arraybuffer' ; this . ws . onopen = this . handleOpen . bind ( this ) ; this . ws . onmessage = ( event ) => { const { data } = event ; if ( isPing ( data ) ) { this . keepAlive ( ) ; this . ws . send ( PONG ) ; } else { const { message , mid } = this . decode ( data ) ; if ( ! message ) return ; if ( typeof mid !== 'number' || mid > MAX_MID || mid < 0 ) { this . emit ( 'data' , message ) ; return ; } if ( this . midsToIgnore . includes ( mid ) ) return ; if ( this . requestedMids . includes ( mid ) ) { this . midsToIgnore . push ( mid ) ; if ( this . midsToIgnore . length > MAX_IGNORE_LEN ) { this . midsToIgnore . splice ( 0 , 1 ) ; } } this . sendAck ( mid ) ; if ( this . lastMid + 1 === mid ) { this . lastMid = mid ; this . emit ( 'data' , message ) ; this . releaseNextRobustMessage ( ) ; return ; } this . robustQueue [ mid ] = message ; const missingMid = this . lastMid + 1 ; this . sendReq ( missingMid ) ; this . requestedMids . push ( missingMid ) ; } } ; this . ws . onerror = ( ) => { if ( this . canConnect === undefined ) { this . canConnect = false ; this . emit ( 'supported' , false ) ; } } ; this . ws . onclose = ( event ) => { const { code , reason } = event ; if ( reason ) { this . canConnect = false ; } this . emit ( 'close' , { code , reason } ) ; if ( this . canConnect ) { if ( this . reconnectAttempts === 0 ) { this . emit ( 'disconnected' ) ; } this . tryReconnect ( ) ; } } ; } send = ( message ) => { if ( this . batchDelay === - 1 ) { if ( this . ws . readyState === this . ws . OPEN ) { this . ws . send ( this . encode ( message ) ) ; } else { this . messageQueue . add ( message ) ; } } else { this . messageQueue . add ( message ) ; if ( ! this . mqTimer ) { this . mqTimer = window . setTimeout ( ( ) => { this . mqTimer = undefined ; const { queue } = this . messageQueue ; if ( this . ws . readyState === this . ws . OPEN && queue . length > 0 ) { this . messageQueue . clear ( ) ; const message = queue . length === 1 ? queue [ 0 ] : queue ; this . ws . send ( this . encode ( message ) ) ; } } , this . batchDelay ) ; } } } ; close ( reason ? ) { this . messageQueue . clear ( ) ; if ( this . ws . readyState === this . ws . CLOSED ) return ; this . ws . close ( 1000 , reason || 'clientClose' ) ; } } export default SocketTrebuchet ;