Skip to content

Commit

Permalink
implement wait() method for QiTx response
Browse files Browse the repository at this point in the history
  • Loading branch information
@alejoacosta74
alejoacosta74 committed Oct 12, 2024
1 parent a2e2d44 commit 5be3d89
Show file tree
Hide file tree
Showing 4 changed files with 158 additions and 17 deletions.
40 changes: 27 additions & 13 deletions src/providers/abstract-provider.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ import {
PollingEventSubscriber,
PollingOrphanSubscriber,
PollingTransactionSubscriber,
QiPollingTransactionSubscriber,
} from './subscriber-polling.js';
import { getNodeLocationFromZone, getZoneFromNodeLocation } from '../utils/shards.js';
import { fromShard } from '../constants/shards.js';
Expand Down Expand Up @@ -159,6 +160,12 @@ export type Subscription =
hash: string;
zone: Zone;
}
| {
type: 'qiTransaction';
tag: string;
hash: string;
zone: Zone;
}
| {
type: 'event';
tag: string;
Expand Down Expand Up @@ -280,7 +287,7 @@ function concisify(items: Array<string>): Array<string> {

// todo `provider` is not used, remove or re-write
// eslint-disable-next-line @typescript-eslint/no-unused-vars
async function getSubscription(_event: ProviderEvent, zone?: Zone): Promise<Subscription> {
async function getSubscription(_event: ProviderEvent, zone?: Zone, hash?: string): Promise<Subscription> {
if (_event == null) {
throw new Error('invalid event');
}
Expand All @@ -307,12 +314,17 @@ async function getSubscription(_event: ProviderEvent, zone?: Zone): Promise<Subs
case 'safe': {
return { type: _event, tag: _event };
}
case 'qiTransaction': {
assert(zone != null, 'zone is required for qiTransaction event', 'MISSING_ARGUMENT');
assert(hash != null, 'hash is required for qiTransaction event', 'MISSING_ARGUMENT');
return { type: _event, tag: _event, hash, zone };
}
}
}

if (isHexString(_event, 32)) {
const hash = _event.toLowerCase();
zone = toZone(hash.slice(0, 4));
zone = zone ?? toZone(hash.slice(0, 4));
return { type: 'transaction', tag: getTag('tx', { hash }), hash, zone };
}

Expand Down Expand Up @@ -1059,7 +1071,7 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
// For QiTransaction, use fromProto() directly
return new QiTransactionResponse(tx, this);
} else {
throw new Error('Unknown transaction type');
throw new Error(`Unknown transaction type: ${tx.type}`);
}
} catch (error) {
console.error('Error in _wrapTransactionResponse:', error);
Expand Down Expand Up @@ -1674,8 +1686,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
return this._wrapBlock(params, network);
}

async getTransaction(hash: string): Promise<null | TransactionResponse> {
const zone = toZone(this.shardFromHash(hash));
async getTransaction(hash: string, _zone?: Zone): Promise<null | TransactionResponse> {
const zone = _zone ?? toZone(this.shardFromHash(hash));
const { network, params } = await resolveProperties({
network: this.getNetwork(),
params: this.#perform({ method: 'getTransaction', hash, zone: zone }),
Expand Down Expand Up @@ -1893,6 +1905,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
return new PollingEventSubscriber(this as AbstractProvider, sub.filter);
case 'transaction':
return new PollingTransactionSubscriber(this as AbstractProvider, sub.hash, sub.zone);
case 'qiTransaction':
return new QiPollingTransactionSubscriber(this as AbstractProvider, sub.hash, sub.zone);
case 'orphan':
return new PollingOrphanSubscriber(this as AbstractProvider, sub.filter, sub.zone);
}
Expand Down Expand Up @@ -1928,8 +1942,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
}
}

async #hasSub(event: ProviderEvent, emitArgs?: Array<any>, zone?: Zone): Promise<null | Sub> {
let sub = await getSubscription(event, zone);
async #hasSub(event: ProviderEvent, emitArgs?: Array<any>, zone?: Zone, hash?: string): Promise<null | Sub> {
let sub = await getSubscription(event, zone, hash);
// This is a log that is removing an existing log; we actually want
// to emit an orphan event for the removed log
if (sub.type === 'event' && emitArgs && emitArgs.length > 0 && emitArgs[0].removed === true) {
Expand All @@ -1938,8 +1952,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
return this.#subs.get(sub.tag) || null;
}

async #getSub(event: ProviderEvent, zone?: Zone): Promise<Sub> {
const subscription = await getSubscription(event, zone);
async #getSub(event: ProviderEvent, zone?: Zone, hash?: string): Promise<Sub> {
const subscription = await getSubscription(event, zone, hash);

// Prevent tampering with our tag in any subclass' _getSubscriber
const tag = subscription.tag;
Expand All @@ -1957,8 +1971,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
return sub;
}

async on(event: ProviderEvent, listener: Listener, zone?: Zone): Promise<this> {
const sub = await this.#getSub(event, zone);
async on(event: ProviderEvent, listener: Listener, zone?: Zone, hash?: string): Promise<this> {
const sub = await this.#getSub(event, zone, hash);
sub.listeners.push({ listener, once: false });
if (!sub.started) {
sub.subscriber.start();
Expand Down Expand Up @@ -2042,8 +2056,8 @@ export class AbstractProvider<C = FetchRequest> implements Provider {
return result;
}

async off(event: ProviderEvent, listener?: Listener, zone?: Zone): Promise<this> {
const sub = await this.#hasSub(event, [], zone);
async off(event: ProviderEvent, listener?: Listener, zone?: Zone, hash?: string): Promise<this> {
const sub = await this.#hasSub(event, [], zone, hash);
if (!sub) {
return this;
}
Expand Down
101 changes: 100 additions & 1 deletion src/providers/provider.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2412,6 +2412,105 @@ export class QiTransactionResponse implements QiTransactionLike, QiTransactionRe
return blockNumber - this.blockNumber + 1;
}

async wait(_confirms?: number, _timeout?: number): Promise<null | QiTransactionResponse> {
const confirms = _confirms == null ? 1 : _confirms;
const timeout = _timeout == null ? 0 : _timeout;

const startBlock = this.startBlock;
let stopScanning = startBlock === -1 ? true : false;

const zoneFromInput = async (txInputs: Array<TxInput> | undefined): Promise<Zone> => {
if (!txInputs || txInputs.length === 0) {
throw new Error('No transaction inputs provided');
}

const firstInput = txInputs[0];
if (!firstInput.pubkey) {
throw new Error('Public key not found in the first transaction input');
}
const address = computeAddress(firstInput.pubkey);
const zone = getZoneForAddress(address);
if (!zone) {
throw new Error(`Invalid zone for address: ${address}`);
}
return zone;
};

const zone = await zoneFromInput(this.txInputs);

const response = await this.provider.getTransaction(this.hash, zone);
if (response && response.isMined() && confirms === 0) {
return response as QiTransactionResponse;
}

if (response) {
if ((await response.confirmations()) >= confirms) {
return response as QiTransactionResponse;
}
} else {
// Allow null only when the confirms is 0
if (confirms === 0) {
return null;
}
}

const waiter = new Promise((resolve, reject) => {
const cancellers: Array<() => void> = [];
const cancel = () => {
cancellers.forEach((c) => c());
};

cancellers.push(() => {
stopScanning = true;
});

if (timeout > 0) {
const timer = setTimeout(() => {
cancel();
reject(makeError('wait for transaction timeout', 'TIMEOUT'));
}, timeout);
cancellers.push(() => {
clearTimeout(timer);
});
}

const txListener = async (response: QiTransactionResponse) => {
if ((await response.confirmations()) >= confirms) {
cancel();
resolve(response);
}
};

cancellers.push(() => {
this.provider.off('qiTransaction', txListener, zone, this.hash);
});
this.provider.on('qiTransaction', txListener, zone, this.hash);

if (startBlock >= 0) {
const blockListener = async () => {
const currentBlock = await this.provider.getBlockNumber(toShard(zone));
if (currentBlock - startBlock >= confirms) {
const response = await this.provider.getTransaction(this.hash, zone);
if (response && response.isMined()) {
cancel();
resolve(response);
}
}

if (!stopScanning) {
this.provider.once('block', blockListener, zone);
}
};
cancellers.push(() => {
this.provider.off('block', blockListener, zone);
});
this.provider.once('block', blockListener, zone);
}
});

return await (<Promise<QiTransactionResponse>>waiter);
}

/**
* Returns `true` if this transaction has been included.
*
Expand Down Expand Up @@ -2860,7 +2959,7 @@ export interface Provider extends ContractRunner, EventEmitterable<ProviderEvent
* @param {string} hash - The transaction hash to fetch.
* @returns {Promise<null | TransactionResponse>} A promise resolving to the transaction or null if not found.
*/
getTransaction(hash: string): Promise<null | TransactionResponse>;
getTransaction(hash: string, zone?: Zone): Promise<null | TransactionResponse>;

/**
* Resolves to the transaction receipt for `hash`, if mined.
Expand Down
30 changes: 29 additions & 1 deletion src/providers/subscriber-polling.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,12 @@ function copy(obj: any): any {
* @returns {Subscriber} The polling subscriber.
* @throws {Error} If the event is unsupported.
*/
export function getPollingSubscriber(provider: AbstractProvider, event: ProviderEvent, zone: Zone): Subscriber {
export function getPollingSubscriber(
provider: AbstractProvider,
event: ProviderEvent,
zone: Zone,
hash?: string,
): Subscriber {
if (event === 'block') {
return new PollingBlockSubscriber(provider, zone);
}
Expand All @@ -34,6 +39,11 @@ export function getPollingSubscriber(provider: AbstractProvider, event: Provider
return new PollingTransactionSubscriber(provider, event, zone);
}

if (event === 'qiTransaction') {
assert(hash != null, "hash is required for 'qiTransaction' event", 'MISSING_ARGUMENT');
return new QiPollingTransactionSubscriber(provider, hash, zone);
}

assert(false, 'unsupported polling event', 'UNSUPPORTED_OPERATION', {
operation: 'getPollingSubscriber',
info: { event },
Expand Down Expand Up @@ -321,6 +331,24 @@ export class PollingTransactionSubscriber extends OnBlockSubscriber {
}
}

export class QiPollingTransactionSubscriber extends OnBlockSubscriber {
#hash: string;
#zone: Zone;

constructor(provider: AbstractProvider, hash: string, zone: Zone) {
super(provider, zone);
this.#hash = hash;
this.#zone = zone;
}

async _poll(blockNumber: number, provider: AbstractProvider): Promise<void> {
const tx = await provider.getTransaction(this.#hash, this.#zone);
if (tx) {
provider.emit(this.#hash, this.#zone, tx);
}
}
}

/**
* A **PollingEventSubscriber** will poll for a given filter for its logs.
*
Expand Down
4 changes: 2 additions & 2 deletions src/utils/events.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ export interface EventEmitterable<T> {
/**
* Registers a `listener` that is called whenever the `event` occurs until unregistered.
*/
on(event: T, listener: Listener, zone?: Zone): Promise<this>;
on(event: T, listener: Listener, zone?: Zone, hash?: string): Promise<this>;

/**
* Registers a `listener` that is called the next time `event` occurs.
Expand All @@ -48,7 +48,7 @@ export interface EventEmitterable<T> {
/**
* Unregister the `listener` for `event`. If `listener` is unspecified, all listeners are unregistered.
*/
off(event: T, listener?: Listener, zone?: Zone): Promise<this>;
off(event: T, listener?: Listener, zone?: Zone, hash?: string): Promise<this>;

/**
* Unregister all listeners for `event`.
Expand Down

0 comments on commit 5be3d89

Please sign in to comment.