fix: race signal instead of setting up listeners (#36)

Ensures listeners are removed as appropriate.

Author: achingbrain

package.json

@@ -66,6 +66,7 @@
     "@multiformats/multiaddr": "^12.4.0",
     "@multiformats/multiaddr-matcher": "^2.0.1",
     "it-stream-types": "^2.0.2",
+    "race-signal": "^1.1.3",
     "uint8arraylist": "^2.4.8"
   },
   "devDependencies": {

src/listener.ts

@@ -1,5 +1,6 @@
 import { setMaxListeners, TypedEventEmitter } from '@libp2p/interface'
 import { multiaddr } from '@multiformats/multiaddr'
+import { raceSignal } from 'race-signal'
 import { QuicConnection } from './connection.js'
 import * as napi from './napi.js'
 import { QuicStreamMuxerFactory } from './stream-muxer.js'
@@ -137,30 +138,27 @@ export class QuicListener extends TypedEventEmitter<ListenerEvents> implements L
     if (this.state.status === 'listening') {
       const signal = this.state.controller.signal
       const listenAddr = this.state.listenAddr
-      const aborted = new Promise((resolve) => {
-        signal.addEventListener('abort', () => { resolve(undefined) }, { once: true })
-      })
+
       while (true) {
         try {
           const listenerPromise = this.state.listener.inboundConnection()
           listenerPromise
             .then(() => this.metrics?.events.increment({ connect: true }))
             .catch(() => this.metrics?.events.increment({ error: true }))
-          const connection = await Promise.race([
-            aborted,
-            listenerPromise
-          ]) as napi.Connection | undefined
-          if (connection == null) {
-            break
-          }
 
+          const connection = await raceSignal(listenerPromise, signal)
           this.onInboundConnection(connection).catch((e) => {
             this.log.error('%s error handling inbound connection', listenAddr.toString(), e)
           })
         } catch (e) {
           this.log.error('%s error accepting connection', listenAddr.toString(), e)
+
+          if (signal.aborted) {
+            break
+          }
         }
       }
+
       this.log('%s no longer awaiting inbound connections', listenAddr.toString())
     }
   }

src/stream-muxer.ts

@@ -1,3 +1,4 @@
+import { raceSignal } from 'race-signal'
 import { QuicStream } from './stream.js'
 import type * as napi from './napi.js'
 import type { AbortOptions, ComponentLogger, Logger, Stream, StreamMuxer, StreamMuxerFactory, StreamMuxerInit } from '@libp2p/interface'
@@ -59,24 +60,20 @@ class QuicStreamMuxer implements StreamMuxer {
   }
 
   async awaitInboundStreams (): Promise<void> {
-    const aborted = new Promise((resolve) => {
-      this.controller.signal.addEventListener('abort', () => { resolve(undefined) }, { once: true })
-    })
     while (true) {
-      const stream = await Promise.race([
-        aborted,
-        this.#connection.inboundStream()
-      ]) as napi.Stream | undefined
-      if (stream == null) {
-        break
-      }
-
       try {
+        const stream = await raceSignal(this.#connection.inboundStream(), this.controller.signal)
+
         this.onInboundStream(stream)
       } catch (e) {
-        this.log.error('%s error accepting stream', this.id, e)
+        this.log.error('%s error accepting stream - %e', this.id, e)
+
+        if (this.controller.signal.aborted) {
+          break
+        }
       }
     }
+
     this.log('%s no longer awaiting inbound streams', this.id)
   }
 
@@ -101,6 +98,7 @@ class QuicStreamMuxer implements StreamMuxer {
 
   async newStream (name?: string): Promise<Stream> {
     const str = await this.#connection.outboundStream()
+    this.controller.signal.throwIfAborted()
     const stream = new QuicStream({
       id: str.id(),
       stream: str,