/**
 * https://github.com/nomic-io/js-tendermint/blob/master/src/rpc.js
 */ import axios from "axios" ; import { EventEmitter } from "events" ; import is from "is_js" ; import ndjson from "ndjson" ; import Pumpify from "pumpify" ; import url from "url" ; import websocket from "websocket-stream" ; export type Args = { [ k ] : any ; } ; function convertHttpArgs ( url , args = { } ) { const search = [ ] ; for ( const k in args ) { if ( is . string ( args [ k ] ) ) { search . push ( `${ k }="${ args [ k ] }"` ) ; } else if ( Buffer . isBuffer ( args [ k ] ) ) { search . push ( `${ k }=0x${ args [ k ] . toString ( "hex" ) }` ) ; } else { search . push ( `${ k }=${ args [ k ] }` ) ; } } return `${ url }?${ search . join ( "&" ) }` ; } function convertWsArgs ( args = { } ) { for ( const k in args ) { const v = args [ k ] ; if ( typeof v === "number" ) { args [ k ] = String ( v ) ; } else if ( Buffer . isBuffer ( v ) ) { args [ k ] = "0x" + v . toString ( "hex" ) ; } else if ( v instanceof Uint8Array ) { args [ k ] = "0x" + Buffer . from ( v ) . toString ( "hex" ) ; } } return args ; } const wsProtocols = [ "ws:" , "wss:" ] ; const httpProtocols = [ "http:" , "https:" ] ; const allProtocols = wsProtocols . concat ( httpProtocols ) ; export default class BaseRpc extends EventEmitter { private uri : string ; public call ! : BaseRpc [ "callWs" ] | BaseRpc [ "callHttp" ] ; private closed = false ; private ws ? : Pumpify ; constructor ( uriString = "localhost:27146" ) { super ( ) ; let { protocol , hostname , port } = url . parse ( uriString ) ; if ( ! protocol || ! allProtocols . includes ( protocol ) ) { const uri = url . parse ( `http://${ uriString }` ) ; protocol = uri . protocol ; hostname = uri . hostname ; port = uri . port ; } this . uri = ! port ? `${ protocol }//${ hostname }/` : `${ protocol }//${ hostname }:${ port }/` ; if ( protocol && wsProtocols . includes ( protocol ) ) { this . uri = `${ this . uri }websocket` ; this . call = this . callWs ; this . connectWs ( ) ; } else if ( protocol && httpProtocols . includes ( protocol ) ) { this . call = this . callHttp ; } } connectWs ( ) { this . ws = new Pumpify . obj ( ndjson . stringify ( ) , websocket ( this . uri ) ) ; this . ws . on ( "error" , ( err ) => this . emit ( "error" , err ) ) ; this . ws . on ( "close" , ( ) => { if ( this . closed ) return ; this . emit ( "error" , Error ( "websocket disconnected" ) ) ; } ) ; this . ws . on ( "data" , ( data ) => { data = JSON . parse ( data ) ; if ( ! data . id ) return ; this . emit ( data . id , data . error , data . result ) ; } ) ; } callHttp ( method , args ? ) { let url = this . uri + method ; url = convertHttpArgs ( url , args ) ; return axios ( { url : url , } ) . then ( function ( { data } ) { if ( data . error ) { const err = Error ( data . error . message ) ; Object . assign ( err , data . error ) ; throw err ; } return data . result ; } , function ( err ) { throw Error ( err ) ; } ) ; } callWs ( method , args ? , listener ? ) { const self = this ; return new Promise ( ( resolve , reject ) => { const id = Math . random ( ) . toString ( 36 ) ; const params = convertWsArgs ( args ) ; if ( method === "subscribe" ) { if ( typeof listener !== "function" ) { throw Error ( "Must provide listener function" ) ; } this . on ( id + "#event" , ( err , res ) => { if ( err ) return self . emit ( "error" , err ) ; return listener ( res . data . value ) ; } ) ; this . on ( id , ( err ) => { if ( err ) return reject ( err ) ; resolve ( ) ; } ) ; } else { this . once ( id , ( err , res ) => { if ( err ) return reject ( err ) ; resolve ( res ) ; } ) ; } this . ws ?. write ( { jsonrpc : "2.0" , id , method , params } ) ; } ) ; } close ( ) { this . closed = true ; if ( ! this . ws ) return ; this . ws . destroy ( ) ; } private createCallBasedMethod = ( name ) => ( args ? , listener ? ) : any => { return this . call ( name , args , listener ) . then ( ( res ) => { return res ; } ) ; } ; subscribe = this . createCallBasedMethod ( "subscribe" ) ; unsubscribe = this . createCallBasedMethod ( "unsubscribe" ) ; unsubscribeAll = this . createCallBasedMethod ( "unsubscribe_all" ) ; status = this . createCallBasedMethod ( "status" ) ; netInfo = this . createCallBasedMethod ( "net_info" ) ; blockchain = this . createCallBasedMethod ( "blockchain" ) ; genesis = this . createCallBasedMethod ( "genesis" ) ; health = this . createCallBasedMethod ( "health" ) ; block = this . createCallBasedMethod ( "block" ) ; blockResults = this . createCallBasedMethod ( "block_results" ) ; validators = this . createCallBasedMethod ( "validators" ) ; consensusState = this . createCallBasedMethod ( "consensus_state" ) ; dumpConsensusState = this . createCallBasedMethod ( "dump_consensus_state" ) ; broadcastTxCommit = this . createCallBasedMethod ( "broadcast_tx_commit" ) ; broadcastTxSync = this . createCallBasedMethod ( "broadcast_tx_sync" ) ; broadcastTxAsync = this . createCallBasedMethod ( "broadcast_tx_async" ) ; unconfirmedTxs = this . createCallBasedMethod ( "unconfirmed_txs" ) ; numUnconfirmedTxs = this . createCallBasedMethod ( "num_unconfirmed_txs" ) ; commit = this . createCallBasedMethod ( "commit" ) ; tx = this . createCallBasedMethod ( "tx" ) ; txSearch = this . createCallBasedMethod ( "tx_search" ) ; abciQuery = this . createCallBasedMethod ( "abci_query" ) ; abciInfo = this . createCallBasedMethod ( "abci_info" ) ; }