feat: allow async stream handlers (#3212)

Allow `await`ing promises inside stream handlers.

Author: achingbrain

packages/interface-compliance-tests/src/mocks/connection.ts

@@ -144,7 +144,7 @@ export function mockConnection (maConn: MultiaddrConnection, opts: MockConnectio
         mss.handle(muxedStream, registrar.getProtocols(), {
           log
         })
-          .then(({ stream, protocol }) => {
+          .then(async ({ stream, protocol }) => {
             log('%s: incoming stream opened on %s', direction, protocol)
             muxedStream.protocol = protocol
             muxedStream.sink = stream.sink
@@ -153,9 +153,10 @@ export function mockConnection (maConn: MultiaddrConnection, opts: MockConnectio
             connection.streams.push(muxedStream)
             const { handler } = registrar.getHandler(protocol)
 
-            handler({ connection, stream: muxedStream })
+            await handler({ connection, stream: muxedStream })
           }).catch(err => {
             log.error(err)
+            muxedStream.abort(err)
           })
       } catch (err: any) {
         log.error(err)

packages/interface/src/stream-handler.ts

@@ -17,7 +17,7 @@ export interface StreamHandler {
   /**
    * A callback function that accepts the incoming stream data
    */
-  (data: IncomingStreamData): void
+  (data: IncomingStreamData): void | Promise<void>
 }
 
 export interface StreamHandlerOptions extends AbortOptions {

packages/libp2p/src/connection.ts

@@ -273,17 +273,12 @@ export class Connection implements ConnectionInterface {
           throw new LimitedConnectionError('Cannot open protocol stream on limited connection')
         }
 
-        handler({ connection: this, stream: muxedStream })
+        await handler({ connection: this, stream: muxedStream })
       })
       .catch(async err => {
         this.log.error('error handling incoming stream id %s - %e', muxedStream.id, err)
 
-        if (muxedStream.timeline.close == null) {
-          await muxedStream.close({
-            signal
-          })
-            .catch(err => muxedStream.abort(err))
-        }
+        muxedStream.abort(err)
       })
   }