fix: accept abort options in connection.newStream (#219)

In order to support aborting during multistream-select, accept an instance of `AbortOptions` as an argument to `connection.newStream`

Author: achingbrain

packages/libp2p-connection/src/index.ts

@@ -5,13 +5,14 @@ import { symbol } from '@libp2p/interfaces/connection'
 import type { Connection, ConnectionStat, Metadata, ProtocolStream, Stream } from '@libp2p/interfaces/connection'
 import type { PeerId } from '@libp2p/interfaces/peer-id'
 import { logger } from '@libp2p/logger'
+import type { AbortOptions } from '@libp2p/interfaces'
 
 const log = logger('libp2p:connection')
 
 interface ConnectionInit {
   remoteAddr: Multiaddr
   remotePeer: PeerId
-  newStream: (protocols: string[]) => Promise<ProtocolStream>
+  newStream: (protocols: string[], options?: AbortOptions) => Promise<ProtocolStream>
   close: () => Promise<void>
   getStreams: () => Stream[]
   stat: ConnectionStat
@@ -47,7 +48,7 @@ export class ConnectionImpl implements Connection {
   /**
    * Reference to the new stream function of the multiplexer
    */
-  private readonly _newStream: (protocols: string[]) => Promise<ProtocolStream>
+  private readonly _newStream: (protocols: string[], options?: AbortOptions) => Promise<ProtocolStream>
   /**
    * Reference to the close function of the raw connection
    */
@@ -102,7 +103,7 @@ export class ConnectionImpl implements Connection {
   /**
    * Create a new stream from this connection
    */
-  async newStream (protocols: string | string[]) {
+  async newStream (protocols: string | string[], options?: AbortOptions) {
     if (this.stat.status === CLOSING) {
       throw errCode(new Error('the connection is being closed'), 'ERR_CONNECTION_BEING_CLOSED')
     }
@@ -115,7 +116,7 @@ export class ConnectionImpl implements Connection {
       protocols = [protocols]
     }
 
-    const { stream, protocol } = await this._newStream(protocols)
+    const { stream, protocol } = await this._newStream(protocols, options)
 
     this.addStream(stream, { protocol, metadata: {} })
 

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

@@ -15,6 +15,7 @@ import * as STATUS from '@libp2p/interfaces/connection/status'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { StreamMuxer } from '@libp2p/interfaces/stream-muxer'
 import { Components } from '@libp2p/interfaces/components'
+import type { AbortOptions } from '@libp2p/interfaces'
 
 const log = logger('libp2p:mock-connection')
 
@@ -65,7 +66,7 @@ class MockConnection implements Connection {
     this.maConn = maConn
   }
 
-  async newStream (protocols: string | string[]) {
+  async newStream (protocols: string | string[], options?: AbortOptions) {
     if (!Array.isArray(protocols)) {
       protocols = [protocols]
     }
@@ -77,7 +78,7 @@ class MockConnection implements Connection {
     const id = `${Math.random()}`
     const stream: Stream = this.muxer.newStream(id)
     const mss = new Dialer(stream)
-    const result = await mss.select(protocols)
+    const result = await mss.select(protocols, options)
 
     const streamData: ProtocolStream = {
       protocol: result.protocol,

packages/libp2p-interfaces/src/connection/index.ts

@@ -3,6 +3,7 @@ import type { PeerId } from '../peer-id'
 import type * as Status from './status.js'
 import type { Duplex } from 'it-stream-types'
 import type { MultiaddrConnection } from '../transport'
+import type { AbortOptions } from '..'
 
 export interface Timeline {
   open: number
@@ -87,7 +88,7 @@ export interface Connection {
   tags: string[]
   streams: Stream[]
 
-  newStream: (multicodecs: string | string[]) => Promise<ProtocolStream>
+  newStream: (multicodecs: string | string[], options?: AbortOptions) => Promise<ProtocolStream>
   addStream: (stream: Stream, data: Partial<Metadata>) => void
   removeStream: (id: string) => void
   close: () => Promise<void>