import mitt from 'mitt' ; import { w3cwebsocket as W3CWebsocket } from 'websocket' ; import { MessageType , NewBlockQuery , NewEventQuery , QueryParam , SocketConnect , SocketState , StatusType , SubscriptionOption , Unsubscribe , } from './types' ; export class WebSocketProvider { public static NewWebSocket ( url , options ? ) { if ( typeof window !== 'undefined' && ( < any > window ) . WebSocket ) { return new WebSocket ( url , options !== undefined ? options . protocol : [ ] ) ; } else { const headers = options !== undefined ? options . headers || { } : undefined ; const urlObject = new URL ( url ) ; if ( headers !== undefined && ! headers . authorization && urlObject . username && urlObject . password ) { const authToken = Buffer . from ( `${ urlObject . username }:${ urlObject . password }` ) . toString ( 'base64' ) ; headers . authorization = `Basic ${ authToken }` ; } return new W3CWebsocket ( url , options !== undefined ? options . protocol : undefined , undefined , headers , undefined , options !== undefined ? options . clientConfig : undefined ) ; } } url : string ; options ? : SubscriptionOption ; emitter : mitt . Emitter ; handlers : any = { } ; websocket : WebSocket | W3CWebsocket ; subscriptions : any ; constructor ( url , options ? ) { this . url = url ; this . options = options ; this . emitter = new mitt ( this . handlers ) ; this . websocket = WebSocketProvider . NewWebSocket ( url , options ) ; this . subscriptions = { } ; this . registerEventListeners ( ) ; } registerEventListeners ( ) { this . websocket . onopen = this . onConnect . bind ( this ) ; this . websocket . onclose = this . onClose . bind ( this ) ; this . websocket . onmessage = this . onMessage . bind ( this ) ; this . websocket . onerror = this . onError . bind ( this ) ; } removeAllSocketListeners ( ) { this . removeEventListener ( SocketState . SOCKET_MESSAGE ) ; this . removeEventListener ( SocketState . SOCKET_READY ) ; this . removeEventListener ( SocketState . SOCKET_CLOSE ) ; this . removeEventListener ( SocketState . SOCKET_ERROR ) ; this . removeEventListener ( SocketState . SOCKET_CONNECT ) ; } removeEventListener ( type ? , handler ? ) { if ( ! type ) { this . handlers = { } ; return ; } if ( ! handler ) { delete this . handlers [ type ] ; } else { return this . emitter . off ( type , handler ) ; } } reconnect ( ) { setTimeout ( ( ) => { this . removeAllSocketListeners ( ) ; this . websocket = WebSocketProvider . NewWebSocket ( this . url , this . options ) ; this . registerEventListeners ( ) ; } , 5000 ) ; } async onClose ( event ) { if ( this . subscriptions !== null && ! event . wasClean ) { this . emitter . emit ( SocketConnect . RECONNECT , event ) ; this . reconnect ( ) ; return ; } if ( this . websocket . CONNECTING ) { this . emitter . emit ( SocketConnect . CLOSE , event ) ; this . websocket . close ( ) ; return ; } } onError ( event ) { this . emitter . emit ( SocketConnect . ERROR , event ) ; if ( this . websocket . CONNECTING ) { this . websocket . close ( ) ; } return ; } async onConnect ( ) { if ( ! this . subscriptions ) { this . subscriptions = { } ; } const subscriptionKeys = Object . keys ( this . subscriptions ) ; if ( subscriptionKeys . length > 0 ) { for ( const key of subscriptionKeys ) { const id = key ; const parameters = this . subscriptions [ key ] . parameters ; delete this . subscriptions [ id ] ; await this . subscribe ( parameters ) ; } } this . emitter . emit ( SocketState . SOCKET_CONNECT ) ; this . emitter . emit ( SocketConnect . CONNECT ) ; } onMessage ( msg ) { if ( msg . data ) { const dataObj = JSON . parse ( msg . data ) ; if ( dataObj . type === MessageType . NOTIFICATION ) { this . emitter . emit ( SocketState . SOCKET_MESSAGE , dataObj ) ; for ( const value of dataObj . values ) { if ( value . query === MessageType . NEW_BLOCK ) { this . emitter . emit ( MessageType . NEW_BLOCK , value ) ; } else if ( value . query === MessageType . EVENT_LOG ) { this . emitter . emit ( MessageType . EVENT_LOG , value ) ; } else if ( value . query === MessageType . UNSUBSCRIBE ) { this . emitter . emit ( MessageType . UNSUBSCRIBE , value ) ; } else { throw new Error ( 'unsupported value type' ) ; } } } else if ( dataObj . query === QueryParam . NEW_BLOCK ) { this . subscriptions [ dataObj . query ] = { id : dataObj . query , parameters : dataObj , } ; this . emitter . emit ( StatusType . SUBSCRIBE_NEW_BLOCK , dataObj ) ; this . emitter . emit ( SocketConnect . RECONNECT ) ; } else if ( dataObj . query === QueryParam . EVENT_LOG ) { this . subscriptions [ dataObj . query ] = { id : dataObj . query , parameters : dataObj , } ; this . emitter . emit ( StatusType . SUBSCRIBE_EVENT_LOG , dataObj ) ; this . emitter . emit ( SocketConnect . RECONNECT ) ; } else if ( dataObj . query === QueryParam . UNSUBSCRIBE ) { this . emitter . emit ( MessageType . UNSUBSCRIBE , dataObj ) ; } else { throw new Error ( 'unsupported message type' ) ; } } else { throw new Error ( 'message data is empty' ) ; } } addEventListener ( type , handler ) { this . emitter . on ( type , handler ) ; } connecting ( ) { return this . websocket . readyState === this . websocket . CONNECTING ; } send ( query ) { return new Promise ( ( resolve , reject ) => { if ( ! this . connecting ( ) ) { try { this . websocket . send ( JSON . stringify ( query ) ) ; } catch ( error ) { throw error ; } let queryParam ; if ( query . query === QueryParam . NEW_BLOCK ) { queryParam = StatusType . SUBSCRIBE_NEW_BLOCK ; } else if ( query . query === QueryParam . EVENT_LOG ) { queryParam = StatusType . SUBSCRIBE_EVENT_LOG ; } else { queryParam = query . query ; } this . emitter . on ( queryParam , ( data ) => { resolve ( data ) ; } ) ; this . emitter . on ( SocketConnect . ERROR , reject ) ; } const connectHandler = ( ) => { this . send ( query ) . then ( resolve ) . catch ( reject ) ; } ; const offConnectHandler = ( ) => { this . emitter . off ( SocketConnect . CONNECT , connectHandler ) ; } ; this . emitter . on ( SocketConnect . CONNECT , connectHandler ) ; this . emitter . on ( SocketConnect . RECONNECT , offConnectHandler ) ; } ) ; } async subscribe ( payload ) { const result = await this . send ( payload ) ; return result . query === payload . query ; } async unsubscribe ( payload ) { const result = await this . send ( payload ) ; const succeed = result . query === payload . query ; if ( succeed ) { this . subscriptions [ payload . query ] = null ; } return succeed ; } }