How to start event listener before calling async method that may trigger the event

[TimoGlastra]

Hi! I have a question on the best way to handle a certain use case with RxJS. I've been using the below method for a while, but wondering if there's a cleaner way to do this. Not that I'm seeing anything wrong with the current approach, but curious if there's other ways to approach this:

Basically I have an async method that will possibly trigger an event, and I want to listen for that event. However I need to start listening for the event before I call the method that may trigger the event. The way I currently approach this is by creating a ReplaySubject of size 1, and subscribe to it beforehand. Then I await the event using firstValueFrom (and add handlers for the case where the event will never arrive)

  private async isBatchPickupSupportedByMediator(mediator: MediationRecord) {
    const { protocolUri } = parseMessageType(BatchPickupMessage.type)

    // Listen for response to our feature query
    const replaySubject = new ReplaySubject(1)
    this.eventEmitter
      .observable<AgentMessageProcessedEvent>(AgentEventTypes.AgentMessageProcessed)
      .pipe(
        // filter by mediator connection id
        filter((e) => e.payload.connection?.id === mediator.connectionId),
        // Search for the query disclose
        filter((e) => e.payload.message.type === DiscloseMessage.type),
        // Map to only the message (and cast type)
        map((e) => e.payload.message as DiscloseMessage),
        // Map to whether the protocol is supported
        map((message) => message.protocols.map((p) => p.protocolId).includes(protocolUri)),
        // TODO: make configurable
        // If we don't have an answer in 7 seconds (no response, not supported, etc...) error
        timeout(7000),
        // We want to return false if an error occurred
        catchError(() => of(false))
      )
      .subscribe(replaySubject)

    // This is the method that may trigger an event
    await this.discoverFeaturesModule.queryFeatures(mediator.connectionId, {
      query: protocolUri,
      comment: 'Detect if batch pickup is supported to determine pickup strategy for messages',
    })

    // Here we await the first value from the replay subject. Which, in this case, will be true or false
    const isBatchPickupSupported = await firstValueFrom(replaySubject)
    return isBatchPickupSupported
  }

I'm wondering where the may be a nice rxjs operator (or something similar) where we start with the event listener being called, but the async method is called directly after so it will be possible to skip the creation of the ReplaySubject?

[voliva]

You don't need to call await on a promise as soon as you have it. You can create a promise, store it in a variable, then await it later.

This would skip the need for a replay subject:

  private async isBatchPickupSupportedByMediator(mediator: MediationRecord) {
    const { protocolUri } = parseMessageType(BatchPickupMessage.type)

    // Listen for response to our feature query
    const response = firstValueFrom(
      this.eventEmitter
        .observable<AgentMessageProcessedEvent>(AgentEventTypes.AgentMessageProcessed)
        .pipe(
          ...
        )
     )

    // This is the method that may trigger an event
    await this.discoverFeaturesModule.queryFeatures(mediator.connectionId, {
      query: protocolUri,
      comment: 'Detect if batch pickup is supported to determine pickup strategy for messages',
    })

    // Here we await the first value from the replay subject. Which, in this case, will be true or false
    // const isBatchPickupSupported = await response
    // return isBatchPickupSupported
    return response; // Returning a promise from an async function is the same as awaiting it.
  }

Because promises are eager, as soon as you call firstValueFrom it will subscribe to the eventEmitter.