Node.js Cross-Chain Bridge Relayers

 
export interface IPooledEventEmitter extends EventEmitter {
 
  /**
   * A unique name to identify it in observability operations.
   */
  name: string;
 
  /**
   * Broadcast an event to all event emitters in the pool.
   * @param eventName The name of the event for event emitters to handle or
   * ignore based on the listeners they have implemented.
   * @param args The arguments associated with the event to pass to the event
   * emitters.
   * @returns An array of event emitter results containing the name of the event
   * emitter and the result it returned. The event name is also returned in the
   * results.
   */
  broadcast(
    eventName: string,
    ...args: unknown[]
  ): BroadcastEventResult;
 
  /**
   * Start all listeners for this event emitter.
   * @returns `true` if listeners started, `false`, if not.
   */
  listen(): boolean;
 
  /**
   * This event emitter can belong to a pool of event emitters. If added to a
   * pool, it can call an event emitter in the pool for further event handling.
   * @param pool The pool instance.
   */
  setPool(pool: IEventEmitterPool): this;
 
  /**
   * Alias for `this.on()`, but checks if the listener exists first before
   * adding it. If a listener already exists for the event, it won't add another one.
   * @param eventName The name of the event to listen for.
   * @param listener The function to call when the event is emitted.
   * @returns `this` instance for further method chaining.
   */
  setOnce(
    eventName: string | symbol,
    listener: (...args: any[]) => any,
  ): this;
}

export class BasePooledEventEmitter<
  Events extends BaseEventMap = BaseEventMap,
> extends EventEmitter<EventMap<Events>> implements IPooledEventEmitter {
  /**
   * This emitter's name for use in key-value mappings in a network.
   */
  readonly name: string;
 
  /**
   * If a caller calls the pool and an error is thrown, then the implementation
   * needs to be fixed so the pool exists.
   */
  protected pool?: IEventEmitterPool;
 
  constructor(
    name: string,
    options?: ConstructorParameters<typeof EventEmitter>[0],
  ) {
    super(
      options ||
        {
          captureRejections:
            process.env.CAPTURE_EVENT_EMITTER_REJECTIONS === "true",
        },
    );
 
    this.name = name;
  }
 
  broadcast(eventName: string, ...args: unknown[]) {
    if (!this.pool) {
      logger.error(
        "This event emitter is not connected to a pool. Cannot send broadcast message.",
      );
 
      return {
        event_name: eventName,
        emitter_results: [],
      };
    }
 
    return this.pool.broadcast(eventName, ...args);
  }
 
  setPool(pool: IEventEmitterPool) {
    logger.debug(`Setting pool: ${pool?.constructor?.name}`);
 
    this.pool = pool;
    return this;
  }
 
  listen() {
    logger.debug("This event emitter does not implement `this.listen()`");
    return false;
  }
 
  /**
   * @alias for `this.on()`, but checks if the listener exists first before
   * adding it.
   */
  setOnce<K>(
    eventName: EventMap<Events> extends [never] ? string | symbol
      : keyof Events | K | "error",
    listener: EventMap<Events> extends [never] ? (...args: any[]) => any
      : K extends keyof Events
        ? EventMap<Events>[K] extends unknown[]
          ? (...args: EventMap<Events>[K]) => any
        : never
      : K extends "error" ? (args: { event: K; error: Error }) => any
      : never,
  ): this {
    // If a listener exists for this event, then skip adding it
    const listeners = this.listeners(eventName);
    if (listeners?.length > 0) {
      logger.debug(`Listener for event '${eventName}' already exists`);
      return this;
    }
 
    logger.debug(`Adding listener for event '${eventName}'`);
 
    // When passing the listener to `this.on()`, its context of `this` gets
    // removed. This causes it to throw errors when it calls `this` in itself.
    // To prevent this, we ensure the listener is bound to `this` so it can call
    // `this` without erroring out.
    const boundListener = listener.bind(this);
 
    // @ts-ignore ...args is untyped right now
    this.on(eventName, async (...args) => {
      try {
        // @ts-ignore ...args is untyped right now
        return await boundListener(...args);
      } catch (e) {
        // @ts-ignore `error` is not added to the event map right now
        this.emit("error", {
          event: eventName,
          error: e,
        });
      }
    });
 
    return this;
  }
 
  override on<K>(
    eventName: EventMap<Events> extends [never] ? string | symbol
      : keyof Events | K,
    listener: EventMap<Events> extends [never] ? (...args: any[]) => void
      : K extends keyof Events ? (...args: EventMap<Events>[K]) => void
      : (...args: any[]) => void,
  ): this {
    logger.debug(`Setting event.on("${eventName}", ...)`);
 
    // @ts-ignore We only care about getting the typing correct for an event
    // name that is created on the fly. Event names created on the fly do not
    // exist in the initial `EventMap` typing for the instance of this class.
    // To get around this type check, we ignore `listener` below since no arg
    // mapping exists for it.
    return super.on(eventName, listener);
  }
 
  override removeAllListeners(
    eventName?:
      | (EventMap<Events> extends [never] ? string | symbol
        : unknown)
      | undefined,
  ): this {
    if (!eventName) {
      return this;
    }
 
    logger.debug(`Removing all listeners for event: ${eventName}`);
 
    return super.removeAllListeners(eventName);
  }
 
  /**
   * Helper method that wraps the starting of all listeners and returns the
   * results of trying to start the listeners.
   * @param cb The callback that should contain all `this.on()` calls.
   * @returns `true` if no errors occurred while starting all the listeners.
   * `false` if an error occurred.
   */
  protected startListeners(cb: () => Promise<void> | void) {
    try {
      cb();
    } catch (error) {
      logger.error(`Failed to start listeners:`, error);
      return false;
    }
 
    return true;
  }
}