refactor(utils): Simple raceForEvent() (#3128)

Implemented a simple `raceForEvent()` utility function which uses
`waitForEvent()` internally. It replaces the previous `raceEvents()`
function which had equal functionality but different implementation for
historical reasons.

## Related changes

Removed unnecessary `executeSafePromise()` wrapper in
`Router#doRouteMessage()`. That code snippet doesn't do async waiting,
so there is no need to have the wrapper.

Author: teogeb

packages/dht/src/connection/connectivityChecker.ts

@@ -1,4 +1,4 @@
-import { Logger, raceEvents } from '@streamr/utils'
+import { Logger, raceForEvent } from '@streamr/utils'
 import { v4 } from 'uuid'
 import * as Err from '../helpers/errors'
 import {
@@ -19,7 +19,7 @@ export const connectAsync = async ({ url, allowSelfSignedCertificate, timeoutMs
     const socket = new WebsocketClientConnection()
     let result: { winnerName: 'connected' | 'error' }
     try {
-        const resultPromise = raceEvents(socket, ['connected', 'error'], timeoutMs)
+        const resultPromise = raceForEvent(socket, ['connected', 'error'], timeoutMs)
         socket.connect(url, allowSelfSignedCertificate)
         result = await resultPromise 
     } catch {

packages/dht/src/dht/routing/Router.ts

@@ -1,6 +1,6 @@
 import { Message, PeerDescriptor, RouteMessageAck, RouteMessageError, RouteMessageWrapper } from '../../../generated/packages/dht/protos/DhtRpc'
 import { RoutingMode, RoutingRemoteContact, RoutingSession } from './RoutingSession'
-import { Logger, executeSafePromise, raceEvents, withTimeout } from '@streamr/utils'
+import { Logger, raceForEvent } from '@streamr/utils'
 import { RoutingRpcCommunicator } from '../../transport/RoutingRpcCommunicator'
 import { DuplicateDetector } from './DuplicateDetector'
 import { v4 } from 'uuid'
@@ -122,22 +122,18 @@ export class Router {
         const contacts = session.updateAndGetRoutablePeers()
         if (contacts.length > 0) {
             this.addRoutingSession(session)
-            logger.trace('starting to raceEvents from routingSession: ' + session.sessionId)
-            let eventReceived: Promise<unknown>
-            executeSafePromise(async () => {
-                eventReceived = raceEvents(
-                    session,
-                    ['routingSucceeded', 'partialSuccess', 'routingFailed', 'stopped'],
-                    null
-                )
-            })
+            logger.trace('starting to raceForEvent from routingSession: ' + session.sessionId)
+            const eventReceived = raceForEvent(
+                session,
+                ['routingSucceeded', 'partialSuccess', 'routingFailed', 'stopped'],
+                10000  // TODO use options option or named constant?
+            )
             setImmediate(async () => {
                 try {
-                    // TODO use options option or named constant?
-                    await withTimeout(eventReceived, 10000)
-                    logger.trace('raceEvents ended from routingSession: ' + session.sessionId)
+                    await eventReceived
+                    logger.trace('raceForEvent ended from routingSession: ' + session.sessionId)
                 } catch {
-                    logger.trace('raceEvents timed out for routingSession ' + session.sessionId) 
+                    logger.trace('raceForEvent timed out for routingSession ' + session.sessionId) 
                 }
                 session.stop()
                 this.removeRoutingSession(session.sessionId) 

packages/utils/src/exports.ts

@@ -25,10 +25,10 @@ export { toEthereumAddressOrENSName } from './toEthereumAddressOrENSName'
 export type { Events, BrandedString } from './types'
 export { wait } from './wait'
 export { waitForEvent } from './waitForEvent'
+export { raceForEvent } from './raceForEvent'
 export { TimeoutError, withTimeout } from './withTimeout'
 export { composeAbortSignals, type ComposedAbortSignal } from './composeAbortSignals'
 export { until } from './until'
-export { raceEvents } from './raceEvents'
 export { withRateLimit } from './withRateLimit'
 export { ObservableEventEmitter } from './ObservableEventEmitter'
 export { initEventGateway } from './initEventGateway'

packages/utils/src/raceEvents.ts

@@ -0,0 +1,32 @@
+import { waitForEvent } from './waitForEvent'
+
+export async function raceForEvent<TEvents extends Record<string, (...args: any[]) => void>, TEventName extends keyof TEvents>(
+    emitter: {
+        on: (eventName: TEventName, listener: TEvents[TEventName]) => unknown
+        off: (eventName: TEventName, listener: TEvents[TEventName]) => unknown
+    },
+    eventNames: TEventName[],
+    timeout: number = 5000
+): Promise<{ winnerName: TEventName, winnerArgs: Parameters<TEvents[TEventName]> }> { 
+    const raceAbortController = new AbortController()
+    const promises = eventNames.map(async (eventName) => {
+        const eventArgs = await waitForEvent(emitter, eventName, timeout, () => true, raceAbortController.signal)
+        return {
+            winnerName: eventName,
+            winnerArgs: eventArgs
+        }
+    })
+    let result
+    try {
+        result = await Promise.race(promises)
+    } finally {
+        // Call raceAbortController.abort() to remove the event listeners. Note that this not strictly needed when Promise.race(promises) rejects.
+        // The race can reject only if withTimeout() timeouts, and as all timeouts happen at the same time, the event listeners are cleaned
+        // up immediately. In that sense this could be moved out from the finally block. But it makes sense to keep it here so that the function
+        // can be seen as an atomic operation: all cleanup happens _before_ the function returns, not immediately _after_ it returns.
+        // The Promise.allSettled() call implements that atomicity by waiting the cleanups initiated by raceAbortController.abort() to complete.
+        raceAbortController.abort()
+        await Promise.allSettled(promises)
+    }
+    return result
+}

packages/utils/src/raceForEvent.ts

@@ -1,13 +1,13 @@
 import { EventEmitter } from 'eventemitter3'
-import { raceEvents } from '../src/raceEvents'
+import { raceForEvent } from '../src/raceForEvent'
 
 interface Events {
     a: (foo: number, bar: number) => void
     b: (foo: string) => void
     c: () => void
 }
 
-describe('raceEvents', () => {
+describe('raceForEvent', () => {
 
     let eventEmitter: EventEmitter<Events>
 
@@ -16,22 +16,22 @@ describe('raceEvents', () => {
     })
 
     it('should resolve with the first event that occurs', async () => {
-        const promise = raceEvents(eventEmitter, ['a', 'b'])
+        const promise = raceForEvent(eventEmitter, ['a', 'b'])
         setTimeout(() => eventEmitter.emit('a', 123, 456), 100)
         setTimeout(() => eventEmitter.emit('b', 'foo'), 10)
         setTimeout(() => eventEmitter.emit('c'), 50)
         expect(await promise).toEqual({ winnerName: 'b', winnerArgs: ['foo'] })
     })
 
     it('should resolve with the correct event and arguments', async () => {
-        const promise = raceEvents(eventEmitter, ['a', 'b'])
+        const promise = raceForEvent(eventEmitter, ['a', 'b'])
         setTimeout(() => eventEmitter.emit('a', 1, 2), 5)
         expect(await promise).toEqual({ winnerName: 'a', winnerArgs: [1, 2] })
     })
 
     it('should clean up listeners after resolving', async () => {
         const offFn = jest.spyOn(eventEmitter, 'off')
-        const promise = raceEvents(eventEmitter, ['a', 'b'])
+        const promise = raceForEvent(eventEmitter, ['a', 'b'])
         eventEmitter.emit('b', 'payload')
         await promise
         expect(offFn).toHaveBeenCalledTimes(2)
@@ -40,7 +40,11 @@ describe('raceEvents', () => {
     })
 
     it('should not resolve if no event is emitted', async () => {
-        const promise = raceEvents(eventEmitter, ['a', 'b'], 50)
+        const offFn = jest.spyOn(eventEmitter, 'off')
+        const promise = raceForEvent(eventEmitter, ['a', 'b'], 50)
         await expect(promise).rejects.toThrow('timed out')
+        expect(offFn).toHaveBeenCalledTimes(2)
+        expect(offFn).toHaveBeenCalledWith('a', expect.any(Function))
+        expect(offFn).toHaveBeenCalledWith('b', expect.any(Function))
     })
 })