"use strict"; import { decode as base64Decode, encode as base64Encode } from "@ethersproject/base64"; import { hexlify, isBytesLike } from "@ethersproject/bytes"; import { shallowCopy } from "@ethersproject/properties"; import { toUtf8Bytes, toUtf8String } from "@ethersproject/strings"; import { Logger } from "@ethersproject/logger"; import { version } from "./_version"; const logger = new Logger(version); import { getUrl, GetUrlResponse, Options } from "./geturl"; function staller(duration: number): Promise { return new Promise((resolve) => { setTimeout(resolve, duration); }); } function bodyify(value: any, type: string): string { if (value == null) { return null; } if (typeof(value) === "string") { return value; } if (isBytesLike(value)) { if (type && (type.split("/")[0] === "text" || type.split(";")[0].trim() === "application/json")) { try { return toUtf8String(value); } catch (error) { }; } return hexlify(value); } return value; } // Exported Types export type ConnectionInfo = { url: string, headers?: { [key: string]: string | number } user?: string, password?: string, allowInsecureAuthentication?: boolean, allowGzip?: boolean, throttleLimit?: number, throttleSlotInterval?: number; throttleCallback?: (attempt: number, url: string) => Promise, skipFetchSetup?: boolean; errorPassThrough?: boolean; timeout?: number, }; export interface OnceBlockable { once(eventName: "block", handler: () => void): void; } export interface OncePollable { once(eventName: "poll", handler: () => void): void; } export type PollOptions = { timeout?: number, floor?: number, ceiling?: number, interval?: number, retryLimit?: number, onceBlock?: OnceBlockable oncePoll?: OncePollable }; export type FetchJsonResponse = { statusCode: number; headers: { [ header: string ]: string }; }; type Header = { key: string, value: string }; // This API is still a work in progress; the future changes will likely be: // - ConnectionInfo => FetchDataRequest // - FetchDataRequest.body? = string | Uint8Array | { contentType: string, data: string | Uint8Array } // - If string => text/plain, Uint8Array => application/octet-stream (if content-type unspecified) // - FetchDataRequest.processFunc = (body: Uint8Array, response: FetchDataResponse) => T // For this reason, it should be considered internal until the API is finalized export function _fetchData(connection: string | ConnectionInfo, body?: Uint8Array, processFunc?: (value: Uint8Array, response: FetchJsonResponse) => T): Promise { // How many times to retry in the event of a throttle const attemptLimit = (typeof(connection) === "object" && connection.throttleLimit != null) ? connection.throttleLimit: 12; logger.assertArgument((attemptLimit > 0 && (attemptLimit % 1) === 0), "invalid connection throttle limit", "connection.throttleLimit", attemptLimit); const throttleCallback = ((typeof(connection) === "object") ? connection.throttleCallback: null); const throttleSlotInterval = ((typeof(connection) === "object" && typeof(connection.throttleSlotInterval) === "number") ? connection.throttleSlotInterval: 100); logger.assertArgument((throttleSlotInterval > 0 && (throttleSlotInterval % 1) === 0), "invalid connection throttle slot interval", "connection.throttleSlotInterval", throttleSlotInterval); const errorPassThrough = ((typeof(connection) === "object") ? !!(connection.errorPassThrough): false); const headers: { [key: string]: Header } = { }; let url: string = null; // @TODO: Allow ConnectionInfo to override some of these values const options: Options = { method: "GET", }; let allow304 = false; let timeout = 2 * 60 * 1000; if (typeof(connection) === "string") { url = connection; } else if (typeof(connection) === "object") { if (connection == null || connection.url == null) { logger.throwArgumentError("missing URL", "connection.url", connection); } url = connection.url; if (typeof(connection.timeout) === "number" && connection.timeout > 0) { timeout = connection.timeout; } if (connection.headers) { for (const key in connection.headers) { headers[key.toLowerCase()] = { key: key, value: String(connection.headers[key]) }; if (["if-none-match", "if-modified-since"].indexOf(key.toLowerCase()) >= 0) { allow304 = true; } } } options.allowGzip = !!connection.allowGzip; if (connection.user != null && connection.password != null) { if (url.substring(0, 6) !== "https:" && connection.allowInsecureAuthentication !== true) { logger.throwError( "basic authentication requires a secure https url", Logger.errors.INVALID_ARGUMENT, { argument: "url", url: url, user: connection.user, password: "[REDACTED]" } ); } const authorization = connection.user + ":" + connection.password; headers["authorization"] = { key: "Authorization", value: "Basic " + base64Encode(toUtf8Bytes(authorization)) }; } if (connection.skipFetchSetup != null) { options.skipFetchSetup = !!connection.skipFetchSetup; } } const reData = new RegExp("^data:([a-z0-9-]+/[a-z0-9-]+);base64,(.*)$", "i"); const dataMatch = ((url) ? url.match(reData): null); if (dataMatch) { try { const response = { statusCode: 200, statusMessage: "OK", headers: { "content-type": dataMatch[1] }, body: base64Decode(dataMatch[2]) }; let result: T = response.body; if (processFunc) { result = processFunc(response.body, response); } return Promise.resolve(result); } catch (error) { logger.throwError("processing response error", Logger.errors.SERVER_ERROR, { body: bodyify(dataMatch[1], dataMatch[2]), error: error, requestBody: null, requestMethod: "GET", url: url }); } } if (body) { options.method = "POST"; options.body = body; if (headers["content-type"] == null) { headers["content-type"] = { key: "Content-Type", value: "application/octet-stream" }; } if (headers["content-length"] == null) { headers["content-length"] = { key: "Content-Length", value: String(body.length) }; } } const flatHeaders: { [ key: string ]: string } = { }; Object.keys(headers).forEach((key) => { const header = headers[key]; flatHeaders[header.key] = header.value; }); options.headers = flatHeaders; const runningTimeout = (function() { let timer: NodeJS.Timer = null; const promise: Promise = new Promise(function(resolve, reject) { if (timeout) { timer = setTimeout(() => { if (timer == null) { return; } timer = null; reject(logger.makeError("timeout", Logger.errors.TIMEOUT, { requestBody: bodyify(options.body, flatHeaders["content-type"]), requestMethod: options.method, timeout: timeout, url: url })); }, timeout); } }); const cancel = function() { if (timer == null) { return; } clearTimeout(timer); timer = null; } return { promise, cancel }; })(); const runningFetch = (async function() { for (let attempt = 0; attempt < attemptLimit; attempt++) { let response: GetUrlResponse = null; try { response = await getUrl(url, options); if (attempt < attemptLimit) { if (response.statusCode === 301 || response.statusCode === 302) { // Redirection; for now we only support absolute locataions const location = response.headers.location || ""; if (options.method === "GET" && location.match(/^https:/)) { url = response.headers.location; continue; } } else if (response.statusCode === 429) { // Exponential back-off throttling let tryAgain = true; if (throttleCallback) { tryAgain = await throttleCallback(attempt, url); } if (tryAgain) { let stall = 0; const retryAfter = response.headers["retry-after"]; if (typeof(retryAfter) === "string" && retryAfter.match(/^[1-9][0-9]*$/)) { stall = parseInt(retryAfter) * 1000; } else { stall = throttleSlotInterval * parseInt(String(Math.random() * Math.pow(2, attempt))); } //console.log("Stalling 429"); await staller(stall); continue; } } } } catch (error) { response = (error).response; if (response == null) { runningTimeout.cancel(); logger.throwError("missing response", Logger.errors.SERVER_ERROR, { requestBody: bodyify(options.body, flatHeaders["content-type"]), requestMethod: options.method, serverError: error, url: url }); } } let body = response.body; if (allow304 && response.statusCode === 304) { body = null; } else if (!errorPassThrough && (response.statusCode < 200 || response.statusCode >= 300)) { runningTimeout.cancel(); logger.throwError("bad response", Logger.errors.SERVER_ERROR, { status: response.statusCode, headers: response.headers, body: bodyify(body, ((response.headers) ? response.headers["content-type"]: null)), requestBody: bodyify(options.body, flatHeaders["content-type"]), requestMethod: options.method, url: url }); } if (processFunc) { try { const result = await processFunc(body, response); runningTimeout.cancel(); return result; } catch (error) { // Allow the processFunc to trigger a throttle if (error.throttleRetry && attempt < attemptLimit) { let tryAgain = true; if (throttleCallback) { tryAgain = await throttleCallback(attempt, url); } if (tryAgain) { const timeout = throttleSlotInterval * parseInt(String(Math.random() * Math.pow(2, attempt))); //console.log("Stalling callback"); await staller(timeout); continue; } } runningTimeout.cancel(); logger.throwError("processing response error", Logger.errors.SERVER_ERROR, { body: bodyify(body, ((response.headers) ? response.headers["content-type"]: null)), error: error, requestBody: bodyify(options.body, flatHeaders["content-type"]), requestMethod: options.method, url: url }); } } runningTimeout.cancel(); // If we had a processFunc, it either returned a T or threw above. // The "body" is now a Uint8Array. return (body); } return logger.throwError("failed response", Logger.errors.SERVER_ERROR, { requestBody: bodyify(options.body, flatHeaders["content-type"]), requestMethod: options.method, url: url }); })(); return Promise.race([ runningTimeout.promise, runningFetch ]); } export function fetchJson(connection: string | ConnectionInfo, json?: string, processFunc?: (value: any, response: FetchJsonResponse) => any): Promise { let processJsonFunc = (value: Uint8Array, response: FetchJsonResponse) => { let result: any = null; if (value != null) { try { result = JSON.parse(toUtf8String(value)); } catch (error) { logger.throwError("invalid JSON", Logger.errors.SERVER_ERROR, { body: value, error: error }); } } if (processFunc) { result = processFunc(result, response); } return result; } // If we have json to send, we must // - add content-type of application/json (unless already overridden) // - convert the json to bytes let body: Uint8Array = null; if (json != null) { body = toUtf8Bytes(json); // Create a connection with the content-type set for JSON const updated: ConnectionInfo = (typeof(connection) === "string") ? ({ url: connection }): shallowCopy(connection); if (updated.headers) { const hasContentType = (Object.keys(updated.headers).filter((k) => (k.toLowerCase() === "content-type")).length) !== 0; if (!hasContentType) { updated.headers = shallowCopy(updated.headers); updated.headers["content-type"] = "application/json"; } } else { updated.headers = { "content-type": "application/json" }; } connection = updated; } return _fetchData(connection, body, processJsonFunc); } export function poll(func: () => Promise, options?: PollOptions): Promise { if (!options) { options = {}; } options = shallowCopy(options); if (options.floor == null) { options.floor = 0; } if (options.ceiling == null) { options.ceiling = 10000; } if (options.interval == null) { options.interval = 250; } return new Promise(function(resolve, reject) { let timer: NodeJS.Timer = null; let done: boolean = false; // Returns true if cancel was successful. Unsuccessful cancel means we're already done. const cancel = (): boolean => { if (done) { return false; } done = true; if (timer) { clearTimeout(timer); } return true; }; if (options.timeout) { timer = setTimeout(() => { if (cancel()) { reject(new Error("timeout")); } }, options.timeout) } const retryLimit = options.retryLimit; let attempt = 0; function check() { return func().then(function(result) { // If we have a result, or are allowed null then we're done if (result !== undefined) { if (cancel()) { resolve(result); } } else if (options.oncePoll) { options.oncePoll.once("poll", check); } else if (options.onceBlock) { options.onceBlock.once("block", check); // Otherwise, exponential back-off (up to 10s) our next request } else if (!done) { attempt++; if (attempt > retryLimit) { if (cancel()) { reject(new Error("retry limit reached")); } return; } let timeout = options.interval * parseInt(String(Math.random() * Math.pow(2, attempt))); if (timeout < options.floor) { timeout = options.floor; } if (timeout > options.ceiling) { timeout = options.ceiling; } setTimeout(check, timeout); } return null; }, function(error) { if (cancel()) { reject(error); } }); } check(); }); }