fix: transport should not handle connection if upgradeInbound throws (#32)

Author: vasco-santos

.aegir.js

@@ -23,8 +23,11 @@ function boot () {
   return listener.listen(ma)
 }
 
-function shutdown () {
-  return listener.close()
+async function shutdown () {
+  await listener.close()
+  // TODO: Temporary fix per wrtc issue
+  // https://github.com/node-webrtc/node-webrtc/issues/636
+  process.exit(0)
 }
 
 module.exports = {

package.json

@@ -46,16 +46,19 @@
     "aegir": "^20.3.1",
     "chai": "^4.2.0",
     "dirty-chai": "^2.0.1",
+    "it-pipe": "^1.1.0",
     "multiaddr": "^7.1.0",
+    "streaming-iterables": "^4.1.1",
     "webrtcsupport": "^2.2.0"
   },
   "dependencies": {
     "abortable-iterator": "^2.1.0",
     "class-is": "^1.1.0",
     "concat-stream": "^2.0.0",
+    "debug": "^4.1.1",
     "detect-node": "^2.0.4",
     "err-code": "^2.0.0",
-    "interface-transport": "libp2p/interface-transport#chore/skip-abort-while-reading-for-webrtc",
+    "libp2p-interfaces": "libp2p/js-interfaces#chore/skip-abort-while-reading-for-webrtc",
     "libp2p-utils": "^0.1.0",
     "mafmt": "^7.0.0",
     "multibase": "~0.6.0",

src/index.js

@@ -21,12 +21,9 @@ const createListener = require('./listener')
 
 function noop () {}
 
-/**
- * @class WebRTCDirect
- */
 class WebRTCDirect {
   /**
-   * @constructor
+   * @class
    * @param {object} options
    * @param {Upgrader} options.upgrader
    */
@@ -36,11 +33,10 @@ class WebRTCDirect {
   }
 
   /**
-   * @async
    * @param {Multiaddr} ma
    * @param {object} options
    * @param {AbortSignal} options.signal Used to abort dial requests
-   * @returns {Connection} An upgraded Connection
+   * @returns {Promise<Connection>} An upgraded Connection
    */
   async dial (ma, options = {}) {
     const socket = await this._connect(ma, options)
@@ -113,6 +109,7 @@ class WebRTCDirect {
         channel.removeListener('error', onError)
         channel.removeListener('timeout', onTimeout)
         channel.removeListener('connect', onConnect)
+        channel.removeAllListeners('signal')
         options.signal && options.signal.removeEventListener('abort', onAbort)
 
         err ? reject(err) : resolve(channel)

src/listener.js

@@ -2,12 +2,13 @@
 
 const http = require('http')
 const EventEmitter = require('events')
-const log = require('debug')('libp2p:webrtcdirect:listener')
+const debug = require('debug')
+const log = debug('libp2p:webrtcdirect:listener')
+log.error = debug('libp2p:webrtcdirect:listener:error')
 
 const isNode = require('detect-node')
 const wrtc = require('wrtc')
 const SimplePeer = require('simple-peer')
-
 const multibase = require('multibase')
 
 const toConnection = require('./socket-to-conn')
@@ -37,26 +38,42 @@ module.exports = ({ handler, upgrader }, options = {}) => {
     }
 
     const channel = new SimplePeer(options)
+
     const maConn = toConnection(channel, listener.__connections)
     log('new inbound connection %s', maConn.remoteAddr)
 
-    const conn = await upgrader.upgradeInbound(maConn)
+    channel.on('error', (err) => {
+      log.error(`incoming connectioned errored with ${err}`)
+    })
+    channel.once('close', () => {
+      channel.removeAllListeners('error')
+    })
+    channel.on('signal', (signal) => {
+      const signalStr = JSON.stringify(signal)
+      const signalEncoded = multibase.encode('base58btc', Buffer.from(signalStr))
+      res.end(signalEncoded.toString())
+    })
+
+    channel.signal(incSignal)
+
+    let conn
+    try {
+      conn = await upgrader.upgradeInbound(maConn)
+    } catch (err) {
+      log.error('inbound connection failed to upgrade', err)
+      return maConn.close()
+    }
     log('inbound connection %s upgraded', maConn.remoteAddr)
 
     trackConn(listener, maConn)
 
     channel.on('connect', () => {
       listener.emit('connection', conn)
       handler(conn)
-    })
 
-    channel.on('signal', (signal) => {
-      const signalStr = JSON.stringify(signal)
-      const signalEncoded = multibase.encode('base58btc', Buffer.from(signalStr))
-      res.end(signalEncoded.toString())
+      channel.removeAllListeners('connect')
+      channel.removeAllListeners('signal')
     })
-
-    channel.signal(incSignal)
   })
 
   server.on('error', (err) => listener.emit('error', err))

test/compliance.js

@@ -1,17 +1,15 @@
 /* eslint-env mocha */
 'use strict'
 
-const wrtc = require('wrtc')
-
-const tests = require('interface-transport')
+const tests = require('libp2p-interfaces/src/transport/tests')
 const multiaddr = require('multiaddr')
 
 const WDirect = require('../src')
 
 describe('interface-transport compliance', () => {
   tests({
     setup ({ upgrader }) {
-      const ws = new WDirect({ upgrader, wrtc: wrtc })
+      const ws = new WDirect({ upgrader })
 
       const addrs = [
         multiaddr('/ip4/127.0.0.1/tcp/22222/http/p2p-webrtc-direct'),