import Trebuchet , { Data , TrebuchetSettings } from './Trebuchet' ; export type FetchData = ( data : any , connectionId : string ) => Promise < Data > ; export type FetchPing = ( connectionId : string ) => Promise < Response > ; export type FetchReliable = ( connectionId : string , data : ArrayBuffer ) => Promise < Response > ; export interface SSESettings extends TrebuchetSettings { getUrl : ( ) => string ; fetchData : FetchData ; fetchPing : FetchPing ; fetchReliable ? : FetchReliable ; } const MAX_INT = 2 ** 31 - 1 ; const MAX_MID = 2 ** 31 - 1 ; const MAX_IGNORE_LEN = 1000 ; const decode = ( msg ) => { const parsedData = JSON . parse ( msg ) ; if ( ! Array . isArray ( parsedData ) ) return { message : parsedData } ; const [ message , mid ] = parsedData ; return { message , mid } ; } ; class SSETrebuchet extends Trebuchet { source ! : EventSource ; private readonly getUrl : ( ) => string ; private readonly fetchData : FetchData ; private readonly fetchPing : FetchPing ; private readonly fetchReliable ? : FetchReliable ; private connectionId : string | undefined = undefined ; constructor ( settings ) { super ( settings ) ; this . getUrl = settings . getUrl ; this . fetchData = settings . fetchData ; this . fetchPing = settings . fetchPing ; this . fetchReliable = settings . fetchReliable ; this . setup ( ) ; } private sendAck ( mid ) { const ack = new Uint8Array ( 4 ) ; const view = new DataView ( ack . buffer ) ; const ackId = mid << 1 ; view . setUint32 ( 0 , ackId , true ) ; this . reply ( 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 . reply ( 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 . source = new EventSource ( this . getUrl ( ) ) ; this . source . onopen = this . handleOpen . bind ( this ) ; this . source . onerror = ( ) => { this . connectionId = undefined ; if ( this . canConnect === undefined ) { this . canConnect = false ; this . source . close ( ) ; this . emit ( 'supported' , false ) ; } else if ( this . canConnect ) { if ( this . reconnectAttempts === 0 ) { this . emit ( 'disconnected' ) ; } this . reconnectAttempts ++ ; } } ; this . source . addEventListener ( 'ka' , ( ) => { if ( ! this . connectionId || ! this . timeout || this . timeout > MAX_INT ) return ; this . fetchPing ( this . connectionId ) . catch ( ) ; clearTimeout ( this . keepAliveTimeoutId ) ; this . keepAliveTimeoutId = window . setTimeout ( ( ) => { this . keepAliveTimeoutId = undefined ; this . source . close ( ) ; this . emit ( 'disconnected' ) ; this . reconnectAttempts ++ ; this . setup ( ) ; } , this . timeout * 1.5 ) ; } ) ; this . source . addEventListener ( 'id' , ( event ) => { this . connectionId = event . data ; this . messageQueue . flush ( this . send ) ; } ) ; this . source . addEventListener ( 'close' , ( event ) => { const splitIdx = event . data . indexOf ( ':' ) ; const code = event . data . slice ( 0 , splitIdx ) ; const reason = event . data . slice ( splitIdx + 1 ) ; this . emit ( 'close' , { code , reason } ) ; this . source . close ( ) ; } ) ; this . source . onmessage = ( event ) => { const { message , mid } = decode ( event . 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 ) ; } ; } ; private handleFetch = async ( message ) => { const res = await this . fetchData ( message , this . connectionId ! ) ; if ( res ) { this . emit ( 'data' , res ) ; } } ; send = ( message ) => { if ( this . source . readyState === this . source . OPEN && this . connectionId ) { this . handleFetch ( message ) . catch ( ) ; } else { this . messageQueue . add ( message ) ; } } ; reply = ( data ) => { if ( this . source . readyState === this . source . OPEN && this . connectionId && this . fetchReliable ) { this . fetchReliable ( this . connectionId , data ) . catch ( ) ; } } ; close ( reason ? ) { this . messageQueue . clear ( ) ; if ( this . source . CLOSED ) return ; this . canConnect = false ; this . source . close ( ) ; this . emit ( 'close' , { code : 1000 , reason } ) ; } } export default SSETrebuchet ;