import { RpcCoder } from '@polkadot/rpc-provider/coder' ; import defaults from '@polkadot/rpc-provider/defaults' ; import { LRUCache } from '@polkadot/rpc-provider/lru' ; import type { JsonRpcResponse , ProviderInterface , ProviderInterfaceCallback , ProviderInterfaceEmitCb , ProviderInterfaceEmitted } from '@polkadot/rpc-provider/types' ; import { getWSErrorString } from '@polkadot/rpc-provider/ws/errors' ; import { assert , isChildClass , isNull , isNumber , isString , isUndefined , objectSpread } from '@polkadot/util' ; import { xglobal } from '@polkadot/x-global' ; import { WebSocket } from '@polkadot/x-ws' ; import chalk from 'chalk' ; import EventEmitter from 'eventemitter3' ; import log from '../logger' ; interface SubscriptionHandler { callback : ProviderInterfaceCallback ; type : string ; } interface WsStateAwaiting { callback : ProviderInterfaceCallback ; method : string ; params : unknown [ ] ; subscription ? : SubscriptionHandler ; } interface WsStateSubscription extends SubscriptionHandler { method : string ; params : unknown [ ] ; } const ALIASES = { chain_finalisedHead : 'chain_finalizedHead' , chain_subscribeFinalisedHeads : 'chain_subscribeFinalizedHeads' , chain_unsubscribeFinalisedHeads : 'chain_unsubscribeFinalizedHeads' } ; const RETRY_DELAY = 2500 ; const MEGABYTE = 1024 * 1024 ; function eraseRecord < T > ( record , cb ? ) { Object . keys ( record ) . forEach ( ( key ) : void => { if ( cb ) { cb ( record [ key ] ) ; } delete record [ key ] ; } ) ; } function shortParams ( result , len ) { try { const str = isString ( result ) || isNumber ( result ) ? result . toString ( ) : JSON . stringify ( result ) ; if ( str . length > len + 3 ) { return `${ str . substr ( 0 , len ) }...` ; } return str ; } catch { return '...' ; } } name description example see export class WsProvider implements ProviderInterface { readonly _callCache = new LRUCache ( ) ; readonly _coder : RpcCoder ; readonly _endpoints : string [ ] ; readonly _headers : Record < string , string > ; readonly _eventemitter : EventEmitter ; readonly _handlers : Record < string , WsStateAwaiting > = { } ; readonly _isReadyPromise : Promise < WsProvider > ; readonly _waitingForId : Record < string , JsonRpcResponse > = { } ; _autoConnectMs : number ; _endpointIndex : number ; _isConnected = false ; _subscriptions : Record < string , WsStateSubscription > = { } ; _websocket : WebSocket | null ; param { string | string [ ] } endpoint param { boolean } autoConnect constructor ( endpoint = defaults . WS_URL , headers = { } ) { const endpoints = Array . isArray ( endpoint ) ? endpoint : [ endpoint ] ; assert ( endpoints . length !== 0 , 'WsProvider requires at least one Endpoint' ) ; endpoints . forEach ( ( endpoint ) => { assert ( /^(wss|ws):\/\// . test ( endpoint ) , ( ) => `Endpoint should start with 'ws://', received '${ endpoint }'` ) ; } ) ; this . _eventemitter = new EventEmitter ( ) ; this . _coder = new RpcCoder ( ) ; this . _endpointIndex = - 1 ; this . _endpoints = endpoints ; this . _headers = headers ; this . _websocket = null ; this . _isReadyPromise = new Promise ( ( resolve ) : void => { this . _eventemitter . once ( 'connected' , ( ) : void => { resolve ( this ) ; } ) ; } ) ; } summary public get hasSubscriptions ( ) : boolean { return true ; } summary return { boolean } public get isConnected ( ) : boolean { return this . _isConnected ; } description public get isReady ( ) : Promise < WsProvider > { return this . _isReadyPromise ; } description public clone ( ) { return new WsProvider ( this . _endpoints ) ; } summary description public async connect ( ) { try { this . _endpointIndex = ( this . _endpointIndex + 1 ) % this . _endpoints . length ; this . _websocket = typeof xglobal . WebSocket !== 'undefined' && isChildClass ( xglobal . WebSocket , WebSocket ) ? new WebSocket ( this . _endpoints [ this . _endpointIndex ] ) : new WebSocket ( this . _endpoints [ this . _endpointIndex ] , undefined , undefined , this . _headers , undefined , { fragmentOutgoingMessages : true , fragmentationThreshold : 1 * MEGABYTE , maxReceivedFrameSize : 24 * MEGABYTE , maxReceivedMessageSize : 24 * MEGABYTE } ) ; this . _websocket . onclose = this . _onSocketClose ; this . _websocket . onerror = this . _onSocketError ; this . _websocket . onmessage = this . _onSocketMessage ; this . _websocket . onopen = this . _onSocketOpen ; return new Promise ( ( resolve , reject ) => { let isConnected = false ; const eventemitter = this . _eventemitter ; function removeAll ( event ? ) { eventemitter . removeListener ( 'connected' , onConnect ) ; eventemitter . removeListener ( 'disconnected' , onClose ) ; eventemitter . removeListener ( 'error' , onClose ) ; if ( isConnected ) { resolve ( ) ; } else { reject ( event ) ; } } function onConnect ( ) { isConnected = true ; removeAll ( ) ; } function onClose ( event ) { isConnected = false ; removeAll ( event ) ; } this . _eventemitter . once ( 'connected' , onConnect ) ; this . _eventemitter . once ( 'disconnected' , onClose ) ; this . _eventemitter . once ( 'error' , onClose ) ; } ) ; } catch ( error ) { log . error ( chalk . red ( error ) ) ; this . _emit ( 'error' , error ) ; throw error ; } } description public async connectWithRetry ( ) { try { await this . connect ( ) ; } catch ( error ) { setTimeout ( ( ) : void => { this . connectWithRetry ( ) . catch ( ( ) : void => { } ) ; } , this . _autoConnectMs || RETRY_DELAY ) ; } } description public async disconnect ( ) { this . _autoConnectMs = 0 ; try { if ( this . _websocket ) { this . _websocket . close ( 1000 ) ; } } catch ( error ) { log . error ( chalk . red ( error ) ) ; this . _emit ( 'error' , error ) ; throw error ; } } summary param { ProviderInterfaceEmitted } type param { ProviderInterfaceEmitCb } sub return public on ( type , sub ) { this . _eventemitter . on ( type , sub ) ; return ( ) : void => { this . _eventemitter . removeListener ( type , sub ) ; } ; } summary param method param params param subscription public send < T = any > ( method , params , isCacheable ? , subscription ? ) { const body = this . _coder . encodeJson ( method , params ) ; let resultPromise = isCacheable ? ( this . _callCache . get ( body ) as Promise < T > ) : null ; if ( ! resultPromise ) { resultPromise = this . _send ( body , method , params , subscription ) ; if ( isCacheable ) { this . _callCache . set ( body , resultPromise ) ; } } return resultPromise ; } async _send < T > ( json , method , params , subscription ? ) { return new Promise < T > ( ( resolve , reject ) : void => { try { assert ( this . isConnected && ! isNull ( this . _websocket ) , 'WebSocket is not connected' ) ; const id = this . _coder . getId ( ) ; const callback = ( error ? , result ? ) : void => { error ? reject ( error ) : resolve ( result as T ) ; } ; log . debug ( `${ chalk . green ( `⬆` ) } Id: ${ chalk . bold ( `%d` ) }, method: ${ method }, params: ${ shortParams ( params , 1000 ) }` , id ) ; this . _handlers [ id ] = { callback , method , params , subscription } ; this . _websocket . send ( json ) ; } catch ( error ) { reject ( error ) ; } } ) ; } name summary example public subscribe ( type , method , params , callback ) { return this . send < number | string > ( method , params , false , { callback , type } ) ; } summary public async unsubscribe ( type , method , id ) { const subscription = `${ type }::${ id }` ; if ( isUndefined ( this . _subscriptions [ subscription ] ) ) { log . warn ( `Unable to find active subscription=${ subscription }` ) ; return false ; } delete this . _subscriptions [ subscription ] ; try { return this . isConnected && ! isNull ( this . _websocket ) ? this . send < boolean > ( method , [ id ] ) : true ; } catch ( error ) { return false ; } } _emit = ( type , ... args ) : void => { this . _eventemitter . emit ( type , ... args ) ; } ; _onSocketClose = ( event ) : void => { const error = new Error ( `disconnected from ${ this . _endpoints [ this . _endpointIndex ] }: ${ event . code }:: ${ event . reason || getWSErrorString ( event . code ) }` ) ; if ( this . _autoConnectMs > 0 ) { log . error ( chalk . red ( error . message ) ) ; } this . _isConnected = false ; if ( this . _websocket ) { this . _websocket . onclose = null ; this . _websocket . onerror = null ; this . _websocket . onmessage = null ; this . _websocket . onopen = null ; this . _websocket = null ; } this . _emit ( 'disconnected' ) ; eraseRecord ( this . _handlers , ( h ) => h . callback ( error , undefined ) ) ; eraseRecord ( this . _waitingForId ) ; if ( this . _autoConnectMs > 0 ) { setTimeout ( ( ) : void => { this . connectWithRetry ( ) . catch ( ( ) => { } ) ; } , this . _autoConnectMs ) ; } } ; _onSocketError = ( error ) : void => { log . warn ( `Socket event: ${ error . type }` ) ; this . _emit ( 'error' , error ) ; } ; _onSocketMessage = ( message ) : void => { const response = JSON . parse ( message . data as string ) as JsonRpcResponse ; const isMsg = isUndefined ( response . method ) ; if ( isMsg ) { log . debug ( `${ chalk . red ( '⬇' ) } Id: ${ chalk . bold ( `%d` ) }, result: ${ shortParams ( response . result , 60 ) }` , response . id ) ; return this . _onSocketMessageResult ( response ) ; } else { log . debug ( `${ chalk . red ( '⬇' ) } SubId: ${ chalk . bold ( response . params . subscription . toString ( ) ) }, result: ${ shortParams ( response . params . result , 60 ) }` ) ; this . _onSocketMessageSubscribe ( response ) ; } } ; _onSocketMessageResult = ( response ) : void => { const handler = this . _handlers [ response . id ] ; if ( ! handler ) { return ; } try { const { method , params , subscription } = handler ; const result = this . _coder . decodeResponse ( response ) as string ; handler . callback ( null , result ) ; if ( subscription ) { const subId = `${ subscription . type }::${ result }` ; this . _subscriptions [ subId ] = objectSpread ( { } , subscription , { method , params } ) ; if ( this . _waitingForId [ subId ] ) { this . _onSocketMessageSubscribe ( this . _waitingForId [ subId ] ) ; } } } catch ( error ) { handler . callback ( error as Error , undefined ) ; } delete this . _handlers [ response . id ] ; } ; _onSocketMessageSubscribe = ( response ) : void => { const method = ALIASES [ response . method as string ] || response . method || 'invalid' ; const subId = `${ method }::${ response . params . subscription }` ; const handler = this . _subscriptions [ subId ] ; if ( ! handler ) { this . _waitingForId [ subId ] = response ; log . warn ( `Unable to find handler for subscription=${ subId }` ) ; return ; } delete this . _waitingForId [ subId ] ; try { const result = this . _coder . decodeResponse ( response ) ; handler . callback ( null , result ) ; } catch ( error ) { handler . callback ( error as Error , undefined ) ; } } ; _onSocketOpen = ( ) : boolean => { assert ( ! isNull ( this . _websocket ) , 'WebSocket cannot be null in onOpen' ) ; log . info ( 'Connected to' , this . _endpoints [ this . _endpointIndex ] ) ; this . _isConnected = true ; this . _emit ( 'connected' ) ; this . _resubscribe ( ) ; return true ; } ; _resubscribe = ( ) : void => { const subscriptions = this . _subscriptions ; this . _subscriptions = { } ; Promise . all ( Object . keys ( subscriptions ) . map ( async ( id ) : Promise < void > => { const { callback , method , params , type } = subscriptions [ id ] ; if ( type . startsWith ( 'author_' ) ) { return ; } try { await this . subscribe ( type , method , params , callback ) ; } catch ( error ) { log . error ( error ) ; } } ) ) . catch ( log . error ) ; } ; }