277 lines
8.0 KiB
TypeScript
277 lines
8.0 KiB
TypeScript
// Local types
|
|
import * as Crypto from 'types/sdk/crypto'
|
|
import { Options } from 'types/sdk/core'
|
|
|
|
// External imports
|
|
import assert from 'assert'
|
|
import { BigNumber } from 'ethers'
|
|
import { bigInt } from 'snarkjs'
|
|
import { randomBytes } from 'crypto'
|
|
|
|
export namespace ErrorUtils {
|
|
export function ensureError<T extends Error>(value: unknown): T {
|
|
if (value instanceof Error) return value as T
|
|
|
|
let stringified = '[Unable to stringify the thrown value]'
|
|
try {
|
|
stringified = JSON.stringify(value)
|
|
} catch {}
|
|
|
|
const error = getError(`This value was thrown as is, not through an Error: ${stringified}`)
|
|
return error as T
|
|
}
|
|
|
|
export function getError(message: string): Error {
|
|
let error = new Error(message)
|
|
error.name = '\nError (tornado-sdk)'
|
|
return error
|
|
}
|
|
}
|
|
|
|
export namespace AsyncUtils {
|
|
class PoolPromise<T> extends Promise<T> {
|
|
orderIndex?: number
|
|
}
|
|
|
|
export type Callback = (...values: any[]) => Promise<any>
|
|
|
|
export type ErrorHandler = (
|
|
err: Error,
|
|
numResolvedPromises: number,
|
|
callbackIndex: number,
|
|
orderIndex: number,
|
|
...args: any[]
|
|
) => void
|
|
|
|
export class PromisePooler {
|
|
concurrencyLimit: number
|
|
private _totalAdded: number
|
|
|
|
_results: Array<any>
|
|
_callbacks: Array<Callback>
|
|
_errorHandlers: Array<ErrorHandler>
|
|
private _promises: Array<PoolPromise<any>>
|
|
|
|
constructor(callbacks: Array<Callback>, errorHandlers: Array<ErrorHandler>, concurrencyLimit: number) {
|
|
if (callbacks.length == 0) throw ErrorUtils.getError('PromisePooler: callbacks are empty.')
|
|
if (concurrencyLimit <= 0)
|
|
throw ErrorUtils.getError("PromisePooler: concurrencyLimit can't be 0 or less.")
|
|
|
|
this.concurrencyLimit = concurrencyLimit
|
|
this._totalAdded = 0
|
|
this._results = []
|
|
this._promises = []
|
|
this._callbacks = callbacks
|
|
this._errorHandlers = errorHandlers
|
|
}
|
|
|
|
get pending(): number {
|
|
return this._promises.length
|
|
}
|
|
|
|
get totalAdded(): number {
|
|
return this._totalAdded
|
|
}
|
|
|
|
async poolMany(...args: Array<Array<any>>): Promise<void> {
|
|
for (const arr of args) {
|
|
await this._waitIfFull()
|
|
this._pool(0, this._totalAdded++, ...arr)
|
|
}
|
|
}
|
|
|
|
async pool(...args: Array<any>): Promise<void> {
|
|
await this._waitIfFull()
|
|
this._pool(0, this._totalAdded++, ...args)
|
|
}
|
|
|
|
async all(): Promise<Array<any>> {
|
|
return Promise.all(this._results.concat(this._promises)).then((result) => {
|
|
this._results = []
|
|
return result
|
|
})
|
|
}
|
|
|
|
async race(): Promise<any> {
|
|
if (this._results.length != 0) {
|
|
return Promise.resolve(this._results.splice(0, 1)[0])
|
|
} else {
|
|
if (this._promises.length != 0) {
|
|
// This will actually return the right value
|
|
// but note that they are stored in results automatically
|
|
// due to the case of 2 promises not being raced in time
|
|
// and as such we instead take from results.
|
|
await Promise.race(this._promises)
|
|
// This should be synchronous
|
|
return this.race()
|
|
} else return Promise.resolve(null)
|
|
}
|
|
}
|
|
|
|
// TODO: Immediately set new callbacks and error handlers
|
|
async reset(): Promise<Array<any>> {
|
|
let results = await this.all()
|
|
|
|
assert(
|
|
this._promises.length === 0,
|
|
ErrorUtils.getError('PromisePooler.reset: Resetting should have allowed all promises to resolve.')
|
|
)
|
|
|
|
this._results = []
|
|
this._totalAdded = 0
|
|
|
|
this._callbacks = []
|
|
this._errorHandlers = []
|
|
|
|
return results
|
|
}
|
|
|
|
private _pool(callbackIndex: number, orderIndex: number, ...args: any[]): Promise<any> {
|
|
let promise: PoolPromise<any> = this._callbacks[callbackIndex](...args).then(
|
|
async (...results) => {
|
|
if (callbackIndex < this._callbacks.length - 1) {
|
|
// This is synchronous, callbackIndex is never 0
|
|
return this._pool(callbackIndex + 1, orderIndex, ...results)
|
|
} else {
|
|
let result = results.length == 1 ? results[0] : results
|
|
this._promises.splice(this._getPromiseIndex(orderIndex), 1)
|
|
this._results.push(result)
|
|
return result
|
|
}
|
|
},
|
|
async (err: Error) => {
|
|
let resolved = this._totalAdded - this.concurrencyLimit
|
|
resolved = resolved < 0 ? 0 : resolved
|
|
// Throw inside to abort
|
|
this._errorHandlers[callbackIndex](
|
|
ErrorUtils.ensureError<Error>(err),
|
|
resolved,
|
|
callbackIndex,
|
|
orderIndex,
|
|
...args
|
|
)
|
|
return this._pool(callbackIndex, orderIndex, ...args)
|
|
}
|
|
)
|
|
|
|
promise.orderIndex = orderIndex
|
|
|
|
const promiseIndex = this._getPromiseIndex(orderIndex)
|
|
|
|
if (promiseIndex < 0) {
|
|
this._promises.push(promise)
|
|
} else {
|
|
this._promises[promiseIndex] = promise
|
|
}
|
|
return promise
|
|
}
|
|
|
|
private _getPromiseIndex(orderIndex: number): number {
|
|
return this._promises.findIndex((_promise) => _promise.orderIndex == orderIndex)
|
|
}
|
|
|
|
private async _waitIfFull(): Promise<void> {
|
|
if (this.concurrencyLimit <= this._promises.length) {
|
|
await Promise.race(this._promises)
|
|
}
|
|
}
|
|
}
|
|
|
|
export class Sync {
|
|
pooler?: PromisePooler
|
|
concurrencyLimit: number
|
|
listen: boolean
|
|
|
|
constructor(options?: Options.Cache.Sync) {
|
|
this.concurrencyLimit = options?.concurrencyLimit ?? 1
|
|
this.listen = options?.listen ?? false
|
|
}
|
|
|
|
initializePooler(callbacks: Array<Callback>, errorHandlers: Array<ErrorHandler>): void {
|
|
if (this.pooler) this.pooler.reset()
|
|
this.pooler = new PromisePooler(callbacks, errorHandlers, this.concurrencyLimit)
|
|
}
|
|
}
|
|
|
|
export function timeout(msTimeout: number): Promise<any> {
|
|
return new Promise((resolve) => setTimeout(resolve, msTimeout))
|
|
}
|
|
}
|
|
|
|
export namespace NumberUtils {
|
|
export function randomBigInteger(numBytes: number): Crypto.bigInt {
|
|
// @ts-ignore
|
|
return bigInt.leBuff2int(randomBytes(numBytes))
|
|
}
|
|
|
|
export function getRandomFromRange(
|
|
lowerInclusive: number,
|
|
upperInclusive: number,
|
|
isInteger: number = 1
|
|
): number {
|
|
isInteger = 0 < isInteger ? 1 : 0
|
|
return (isInteger ? Math.floor : (x: number) => x)(
|
|
Math.random() * (upperInclusive + isInteger - lowerInclusive) + lowerInclusive
|
|
)
|
|
}
|
|
}
|
|
|
|
export namespace HexUtils {
|
|
export function bufferToHex(buffer: Buffer, byteLen: number = 32): string {
|
|
return '0x' + buffer.toString('hex').padStart(byteLen * 2, '0')
|
|
}
|
|
|
|
export function numberToHex(number: number): string {
|
|
return BigNumber.from(number).toHexString()
|
|
}
|
|
|
|
export function bigIntToHex(number: Crypto.bigInt, byteLen: number = 32): string {
|
|
// @ts-ignore
|
|
return '0x' + number.toString(16).padStart(2 * byteLen, '0')
|
|
}
|
|
|
|
export function prepareAddress(address: string, bytelen: number = 32): string {
|
|
return (address.slice(0, 2) == '0x' ? address.slice(2) : address).toLowerCase().padStart(bytelen * 2, '0')
|
|
}
|
|
}
|
|
|
|
export namespace ObjectUtils {
|
|
export function populate<T extends Object>(source?: T, target?: T): T {
|
|
if (!source) {
|
|
if (!target) {
|
|
throw ErrorUtils.getError('ObjectUtils.populate: source & target object null or undefined')
|
|
} else return target
|
|
}
|
|
if (!target) throw ErrorUtils.getError('ObjectUtils.populate: target object null or undefined')
|
|
|
|
type KeyType = keyof typeof target
|
|
const sourceEntries: [string, any][] = Object.entries(source)
|
|
|
|
sourceEntries.forEach((sEntry: [string, any]) => {
|
|
const key = sEntry[0] as KeyType
|
|
// Get entry in target
|
|
const tEntry = target[key]
|
|
// Set if it's not present in the target
|
|
if (tEntry === undefined || tEntry === null) target[key] = sEntry[1]
|
|
})
|
|
|
|
return target
|
|
}
|
|
|
|
type Swapped<T extends Object> = {
|
|
left: T
|
|
right: T
|
|
}
|
|
|
|
export function swap<T extends Object>(left: T, right: T): Swapped<T> {
|
|
return {
|
|
left: right,
|
|
right: left
|
|
}
|
|
}
|
|
|
|
export function exists(obj: any): boolean {
|
|
return obj !== undefined && obj !== null
|
|
}
|
|
}
|