import { assert, isHexString } from "../utils/index.js"; import type { AbstractProvider, Subscriber } from "./abstract-provider.js"; import type { EventFilter, OrphanFilter, ProviderEvent } from "./provider.js"; function copy(obj: any): any { return JSON.parse(JSON.stringify(obj)); } /** * Return the polling subscriber for common events. * * @_docloc: api/providers/abstract-provider */ export function getPollingSubscriber(provider: AbstractProvider, event: ProviderEvent): Subscriber { if (event === "block") { return new PollingBlockSubscriber(provider); } if (isHexString(event, 32)) { return new PollingTransactionSubscriber(provider, event); } assert(false, "unsupported polling event", "UNSUPPORTED_OPERATION", { operation: "getPollingSubscriber", info: { event } }); } // @TODO: refactor this /** * A **PollingBlockSubscriber** polls at a regular interval for a change * in the block number. * * @_docloc: api/providers/abstract-provider */ export class PollingBlockSubscriber implements Subscriber { #provider: AbstractProvider; #poller: null | number; #interval: number; // The most recent block we have scanned for events. The value -2 // indicates we still need to fetch an initial block number #blockNumber: number; /** * Create a new **PollingBlockSubscriber** attached to %%provider%%. */ constructor(provider: AbstractProvider) { this.#provider = provider; this.#poller = null; this.#interval = 4000; this.#blockNumber = -2; } /** * The polling interval. */ get pollingInterval(): number { return this.#interval; } set pollingInterval(value: number) { this.#interval = value; } async #poll(): Promise { try { const blockNumber = await this.#provider.getBlockNumber(); // Bootstrap poll to setup our initial block number if (this.#blockNumber === -2) { this.#blockNumber = blockNumber; return; } // @TODO: Put a cap on the maximum number of events per loop? if (blockNumber !== this.#blockNumber) { for (let b = this.#blockNumber + 1; b <= blockNumber; b++) { // We have been stopped if (this.#poller == null) { return; } await this.#provider.emit("block", b); } this.#blockNumber = blockNumber; } } catch (error) { // @TODO: Minor bump, add an "error" event to let subscribers // know things went awry. //console.log(error); } // We have been stopped if (this.#poller == null) { return; } this.#poller = this.#provider._setTimeout(this.#poll.bind(this), this.#interval); } start(): void { if (this.#poller) { return; } this.#poller = this.#provider._setTimeout(this.#poll.bind(this), this.#interval); this.#poll(); } stop(): void { if (!this.#poller) { return; } this.#provider._clearTimeout(this.#poller); this.#poller = null; } pause(dropWhilePaused?: boolean): void { this.stop(); if (dropWhilePaused) { this.#blockNumber = -2; } } resume(): void { this.start(); } } /** * An **OnBlockSubscriber** can be sub-classed, with a [[_poll]] * implmentation which will be called on every new block. * * @_docloc: api/providers/abstract-provider */ export class OnBlockSubscriber implements Subscriber { #provider: AbstractProvider; #poll: (b: number) => void; #running: boolean; /** * Create a new **OnBlockSubscriber** attached to %%provider%%. */ constructor(provider: AbstractProvider) { this.#provider = provider; this.#running = false; this.#poll = (blockNumber: number) => { this._poll(blockNumber, this.#provider); } } /** * Called on every new block. */ async _poll(blockNumber: number, provider: AbstractProvider): Promise { throw new Error("sub-classes must override this"); } start(): void { if (this.#running) { return; } this.#running = true; this.#poll(-2); this.#provider.on("block", this.#poll); } stop(): void { if (!this.#running) { return; } this.#running = false; this.#provider.off("block", this.#poll); } pause(dropWhilePaused?: boolean): void { this.stop(); } resume(): void { this.start(); } } export class PollingBlockTagSubscriber extends OnBlockSubscriber { readonly #tag: string; #lastBlock: number; constructor(provider: AbstractProvider, tag: string) { super(provider); this.#tag = tag; this.#lastBlock = -2; } pause(dropWhilePaused?: boolean): void { if (dropWhilePaused) { this.#lastBlock = -2; } super.pause(dropWhilePaused); } async _poll(blockNumber: number, provider: AbstractProvider): Promise { const block = await provider.getBlock(this.#tag); if (block == null) { return; } if (this.#lastBlock === -2) { this.#lastBlock = block.number; } else if (block.number > this.#lastBlock) { provider.emit(this.#tag, block.number); this.#lastBlock = block.number; } } } /** * @_ignore: * * @_docloc: api/providers/abstract-provider */ export class PollingOrphanSubscriber extends OnBlockSubscriber { #filter: OrphanFilter; constructor(provider: AbstractProvider, filter: OrphanFilter) { super(provider); this.#filter = copy(filter); } async _poll(blockNumber: number, provider: AbstractProvider): Promise { throw new Error("@TODO"); console.log(this.#filter); } } /** * A **PollingTransactionSubscriber** will poll for a given transaction * hash for its receipt. * * @_docloc: api/providers/abstract-provider */ export class PollingTransactionSubscriber extends OnBlockSubscriber { #hash: string; /** * Create a new **PollingTransactionSubscriber** attached to * %%provider%%, listening for %%hash%%. */ constructor(provider: AbstractProvider, hash: string) { super(provider); this.#hash = hash; } async _poll(blockNumber: number, provider: AbstractProvider): Promise { const tx = await provider.getTransactionReceipt(this.#hash); if (tx) { provider.emit(this.#hash, tx); } } } /** * A **PollingEventSubscriber** will poll for a given filter for its logs. * * @_docloc: api/providers/abstract-provider */ export class PollingEventSubscriber implements Subscriber { #provider: AbstractProvider; #filter: EventFilter; #poller: (b: number) => void; #running: boolean; // The most recent block we have scanned for events. The value -2 // indicates we still need to fetch an initial block number #blockNumber: number; /** * Create a new **PollingTransactionSubscriber** attached to * %%provider%%, listening for %%filter%%. */ constructor(provider: AbstractProvider, filter: EventFilter) { this.#provider = provider; this.#filter = copy(filter); this.#poller = this.#poll.bind(this); this.#running = false; this.#blockNumber = -2; } async #poll(blockNumber: number): Promise { // The initial block hasn't been determined yet if (this.#blockNumber === -2) { return; } const filter = copy(this.#filter); filter.fromBlock = this.#blockNumber + 1; filter.toBlock = blockNumber; const logs = await this.#provider.getLogs(filter); // No logs could just mean the node has not indexed them yet, // so we keep a sliding window of 60 blocks to keep scanning if (logs.length === 0) { if (this.#blockNumber < blockNumber - 60) { this.#blockNumber = blockNumber - 60; } return; } for (const log of logs) { this.#provider.emit(this.#filter, log); // Only advance the block number when logs were found to // account for networks (like BNB and Polygon) which may // sacrifice event consistency for block event speed this.#blockNumber = log.blockNumber; } } start(): void { if (this.#running) { return; } this.#running = true; if (this.#blockNumber === -2) { this.#provider.getBlockNumber().then((blockNumber) => { this.#blockNumber = blockNumber; }); } this.#provider.on("block", this.#poller); } stop(): void { if (!this.#running) { return; } this.#running = false; this.#provider.off("block", this.#poller); } pause(dropWhilePaused?: boolean): void { this.stop(); if (dropWhilePaused) { this.#blockNumber = -2; } } resume(): void { this.start(); } }