import { Span , SpanKind , tracer } from '@neo-one/client-switch' ; import { addAttributesToSpan , Labels } from '@neo-one/utils' ; import _fetch from 'cross-fetch' ; import DataLoader from 'dataloader' ; import debug from 'debug' ; import stringify from 'safe-stable-stringify' ; import { HTTPError , InvalidRPCResponseError , JSONRPCError } from '../errors' ; import { AbortController } from './AbortController.ponyfill' ; import { JSONRPCProvider , JSONRPCRequest } from './JSONRPCProvider' ; const logger = debug ( 'NEOONE:JSONRPCProvider' ) ; const TIMEOUT_MS = 20000 ; const WATCH_TIMEOUT_MS = 5000 ; const PARSE_ERROR_CODE = - 32700 ; const PARSE_ERROR_MESSAGE = 'Parse error' ; const getWaitTime = ( response ) => { const resetTimeout = response . headers . get ( 'Retry-After' ) ; return resetTimeout !== null ? Math . max ( Number ( resetTimeout ) , 1 ) + 2 : 2 ; } ; const browserFetch = async ( input , init , timeoutMS ) : Promise < Response > => { const controller = new AbortController ( ) ; const responsePromise = _fetch ( input , { ... init , signal : controller . signal as any } ) ; const timeout = setTimeout ( ( ) => controller . abort ( ) , timeoutMS ) ; try { const response = await responsePromise ; return response ; } finally { clearTimeout ( timeout ) ; } } ; const nodeFetch = async ( input , init , timeoutMS ) : Promise < Response > => _fetch ( input , { ... init , timeout : timeoutMS } as any ) ; const fetch = typeof window === 'undefined' ? nodeFetch : browserFetch ; const instrumentFetch = async < T extends { readonly status : number ; } > ( doFetch , endpoint , type , span ) => { const headers = { 'Content-Type' : 'application/json' } ; const labels = { [ Labels . HTTP_URL ] : endpoint , [ Labels . HTTP_METHOD ] : 'POST' , [ Labels . JSONRPC_TYPE ] : type } ; addAttributesToSpan ( span , labels ) ; try { if ( tracer . active ) { tracer . propagation . inject ( { setHeader : ( name , value ) => { headers [ name ] = value ; } } , span . spanContext ) ; } let status = - 1 ; try { const resp = await doFetch ( headers ) ; status = resp . status ; logger ( '%o' , { level : 'debug' , name : 'http_client_request' , ... labels } ) ; return resp ; } finally { span . addAttribute ( Labels . HTTP_STATUS_CODE , status ) ; } } catch ( error ) { logger ( '%o' , { level : 'error' , name : 'http_client_request' , ... labels , error : error . message } ) ; throw error ; } } ; const doRequest = async ( { endpoint , requests , timeoutMS , tries , } ) => { const body = JSON . stringify ( requests . map ( ( req ) => req . request ) ) ; const span = requests [ 0 ] . span ; let remainingTries = tries ; let parseErrorTries = 3 ; let rateLimitTimeout ; let result ; let finalError ; while ( remainingTries >= 0 ) { try { if ( rateLimitTimeout !== undefined ) { const sleepTime = rateLimitTimeout ; rateLimitTimeout = undefined ; finalError = undefined ; await new Promise < void > ( ( resolve ) => setTimeout ( resolve , sleepTime * 1000 ) ) ; } const response = await instrumentFetch ( async ( headers ) => fetch ( endpoint , { method : 'POST' , headers , body } , timeoutMS ) , endpoint , 'fetch' , span ) ; if ( ! response . ok ) { let text ; try { text = await response . text ( ) ; } catch { } if ( response . status === 429 ) { rateLimitTimeout = getWaitTime ( response ) ; } throw new HTTPError ( response . status , text ) ; } result = await response . json ( ) ; if ( Array . isArray ( result ) ) { return result ; } if ( typeof result === 'object' && result . error !== undefined && typeof result . error === 'object' && typeof result . error . code === 'number' && typeof result . error . message === 'string' ) { if ( result . error . code === PARSE_ERROR_CODE && result . error . message === PARSE_ERROR_MESSAGE && parseErrorTries > 0 ) { remainingTries += 1 ; parseErrorTries -= 1 ; } else { throw new JSONRPCError ( result . error ) ; } } } catch ( error ) { finalError = error ; } remainingTries -= 1 ; } if ( finalError !== undefined ) { throw finalError ; } throw new InvalidRPCResponseError ( ) ; } ; const watchSingle = async ( { endpoint , req , timeoutMS , span , } ) : Promise < any > => { const response = await instrumentFetch ( async ( headers ) => fetch ( endpoint , { method : 'POST' , headers , body : JSON . stringify ( req ) } , timeoutMS + WATCH_TIMEOUT_MS ) , endpoint , 'watch' , span ) ; if ( ! response . ok ) { let text ; try { text = await response . text ( ) ; } catch { } if ( response . status === 429 ) { await new Promise < void > ( ( resolve ) => setTimeout ( resolve , getWaitTime ( response ) * 1000 ) ) ; return watchSingle ( { endpoint , req , timeoutMS , span } ) ; } throw new HTTPError ( response . status , text ) ; } return response . json ( ) ; } ; /**
 * Implements the `JSONRPCProvider` interface using http requests.
 */ export class JSONRPCHTTPProvider extends JSONRPCProvider { public readonly endpoint : string ; public readonly batcher : DataLoader < { readonly span : Span ; readonly request : any ; } , any > ; public constructor ( endpoint ) { super ( ) ; this . endpoint = endpoint ; this . batcher = new DataLoader ( async ( requests ) => { this . batcher . clearAll ( ) ; return doRequest ( { endpoint , requests , tries : 1 , timeoutMS : TIMEOUT_MS } ) ; } , { maxBatchSize : 25 , cacheKeyFn : ( value ) => stringify ( value . request ) } ) ; } public async request ( req ) { const labels = { [ Labels . RPC_TYPE ] : 'jsonrpc' , [ Labels . RPC_METHOD ] : req . method , [ Labels . SPAN_KIND ] : 'client' } ; return tracer . startRootSpan ( { name : 'json_rpc_client_request' , kind : SpanKind . CLIENT } , async ( span ) => { addAttributesToSpan ( span , labels ) ; try { const result = await this . requestInternal ( req , span ) ; logger ( { level : 'debug' , name : 'jsonrpc_client_request' , ... labels } ) ; return result ; } catch ( error ) { logger ( { level : 'error' , name : 'jsonrpc_client_request' , ... labels , error : error . message } ) ; throw error ; } finally { span . end ( ) ; } } ) ; } private async requestInternal ( req , span ) { let response ; const { watchTimeoutMS , params = [ ] } = req ; if ( watchTimeoutMS !== undefined ) { response = await watchSingle ( { endpoint : this . endpoint , req : { jsonrpc : '2.0' , id : 1 , method : req . method , params : params . concat ( [ watchTimeoutMS ] ) } , span , timeoutMS : watchTimeoutMS } ) ; } else { response = await this . batcher . load ( { span , request : { jsonrpc : '2.0' , id : 1 , method : req . method , params } } ) ; } return this . handleResponse ( response ) ; } }