refactor(async): add dialer and upgrader

package.json

@@ -41,22 +41,25 @@
     "npm": ">=6.0.0"
   },
   "dependencies": {
+    "abort-controller": "^3.0.0",
     "async": "^2.6.2",
     "bignumber.js": "^9.0.0",
     "class-is": "^1.1.0",
     "debug": "^4.1.1",
     "err-code": "^1.1.2",
     "fsm-event": "^2.1.0",
     "hashlru": "^2.3.0",
-    "interface-connection": "~0.3.3",
+    "it-pipe": "^1.0.1",
     "latency-monitor": "~0.2.1",
     "libp2p-crypto": "^0.16.2",
+    "libp2p-interfaces": "^0.1.1",
     "mafmt": "^7.0.0",
     "merge-options": "^1.0.1",
     "moving-average": "^1.0.0",
     "multiaddr": "^7.1.0",
-    "multistream-select": "~0.14.6",
+    "multistream-select": "^0.15.0",
     "once": "^1.4.0",
+    "p-queue": "^6.1.1",
     "p-settle": "^3.1.0",
     "peer-book": "^0.9.1",
     "peer-id": "^0.13.3",
@@ -73,29 +76,33 @@
   },
   "devDependencies": {
     "@nodeutils/defaults-deep": "^1.1.0",
+    "abortable-iterator": "^2.1.0",
     "aegir": "^20.0.0",
     "chai": "^4.2.0",
     "chai-checkmark": "^1.0.1",
     "cids": "^0.7.1",
     "delay": "^4.3.0",
     "dirty-chai": "^2.0.1",
     "electron-webrtc": "^0.3.0",
+    "glob": "^7.1.4",
     "interface-datastore": "^0.6.0",
+    "it-pair": "^1.0.0",
     "libp2p-bootstrap": "^0.9.7",
     "libp2p-delegated-content-routing": "^0.2.2",
     "libp2p-delegated-peer-routing": "^0.2.2",
     "libp2p-floodsub": "~0.17.0",
     "libp2p-gossipsub": "~0.0.4",
     "libp2p-kad-dht": "^0.15.3",
     "libp2p-mdns": "^0.12.3",
-    "libp2p-mplex": "^0.8.4",
+    "libp2p-mplex": "^0.9.1",
     "libp2p-pnet": "~0.1.0",
     "libp2p-secio": "^0.11.1",
     "libp2p-spdy": "^0.13.2",
     "libp2p-tcp": "^0.14.1",
     "libp2p-websockets": "^0.13.0",
     "lodash.times": "^4.3.2",
     "nock": "^10.0.6",
+    "p-defer": "^3.0.0",
     "portfinder": "^1.0.20",
     "pull-goodbye": "0.0.2",
     "pull-length-prefixed": "^1.3.3",
@@ -104,6 +111,7 @@
     "pull-protocol-buffers": "~0.1.2",
     "pull-serializer": "^0.3.2",
     "sinon": "^7.2.7",
+    "streaming-iterables": "^4.1.0",
     "wrtc": "^0.4.1"
   },
   "contributors": [

src/circuit/README.md

@@ -1,6 +1,6 @@
 # js-libp2p-circuit
 
-> Node.js implementation of the Circuit module that libp2p uses, which implements the [interface-connection](https://github.com/libp2p/interface-connection) interface for dial/listen.
+> Node.js implementation of the Circuit module that libp2p uses, which implements the [interface-connection](https://github.com/libp2p/js-interfaces/tree/master/src/connection) interface for dial/listen.
 
 **Note**: git history prior to merging into js-libp2p can be found in the original repository, https://github.com/libp2p/js-libp2p-circuit.
 
@@ -24,15 +24,18 @@ Prior to `libp2p-circuit` there was a rift in the IPFS network, were IPFS nodes
 
 ## Table of Contents
 
-- [Install](#install)
-  - [npm](#npm)
-- [Usage](#usage)
-  - [Example](#example)
-  - [This module uses `pull-streams`](#this-module-uses-pull-streams)
-    - [Converting `pull-streams` to Node.js Streams](#converting-pull-streams-to-nodejs-streams)
-- [API](#api)
-- [Contribute](#contribute)
-- [License](#license)
+- [js-libp2p-circuit](#js-libp2p-circuit)
+    - [Why?](#why)
+    - [libp2p-circuit and IPFS](#libp2p-circuit-and-ipfs)
+  - [Table of Contents](#table-of-contents)
+  - [Usage](#usage)
+    - [Example](#example)
+      - [Create dialer/listener](#create-dialerlistener)
+      - [Create `relay`](#create-relay)
+    - [This module uses `pull-streams`](#this-module-uses-pull-streams)
+      - [Converting `pull-streams` to Node.js Streams](#converting-pull-streams-to-nodejs-streams)
+  - [API](#api)
+    - [Implementation rational](#implementation-rational)
 
 ## Usage
 

src/circuit/circuit/dialer.js

@@ -6,7 +6,7 @@ const waterfall = require('async/waterfall')
 const setImmediate = require('async/setImmediate')
 const multiaddr = require('multiaddr')
 
-const Connection = require('interface-connection').Connection
+const { Connection } = require('libp2p-interfaces/src/connection')
 
 const utilsFactory = require('./utils')
 const StreamHandler = require('./stream-handler')

src/circuit/circuit/stop.js

@@ -3,7 +3,7 @@
 const setImmediate = require('async/setImmediate')
 
 const EE = require('events').EventEmitter
-const Connection = require('interface-connection').Connection
+const { Connection } = require('libp2p-interfaces/src/connection')
 const utilsFactory = require('./utils')
 const PeerInfo = require('peer-info')
 const proto = require('../protocol').CircuitRelay

src/constants.js

@@ -0,0 +1,12 @@
+'use strict'
+
+module.exports = {
+  DENY_TTL: 5 * 60 * 1e3, // How long before an errored peer can be dialed again
+  DENY_ATTEMPTS: 5, // Num of unsuccessful dials before a peer is permanently denied
+  DIAL_TIMEOUT: 30e3, // How long in ms a dial attempt is allowed to take
+  MAX_COLD_CALLS: 50, // How many dials w/o protocols that can be queued
+  MAX_PARALLEL_DIALS: 100, // Maximum allowed concurrent dials
+  QUARTER_HOUR: 15 * 60e3,
+  PRIORITY_HIGH: 10,
+  PRIORITY_LOW: 20
+}

src/dialer.js

@@ -0,0 +1,98 @@
+'use strict'
+
+const multiaddr = require('multiaddr')
+const errCode = require('err-code')
+const { default: PQueue } = require('p-queue')
+const AbortController = require('abort-controller')
+const debug = require('debug')
+const log = debug('libp2p:dialer')
+log.error = debug('libp2p:dialer:error')
+
+const { codes } = require('./errors')
+const {
+  MAX_PARALLEL_DIALS,
+  DIAL_TIMEOUT
+} = require('./constants')
+
+class Dialer {
+  /**
+   * @constructor
+   * @param {object} options
+   * @param {TransportManager} options.transportManager
+   * @param {number} options.concurrency Number of max concurrent dials. Defaults to `MAX_PARALLEL_DIALS`
+   * @param {number} options.timeout How long a dial attempt is allowed to take. Defaults to `DIAL_TIMEOUT`
+   */
+  constructor ({
+    transportManager,
+    concurrency = MAX_PARALLEL_DIALS,
+    timeout = DIAL_TIMEOUT
+  }) {
+    this.transportManager = transportManager
+    this.concurrency = concurrency
+    this.timeout = timeout
+    this.queue = new PQueue({ concurrency, timeout, throwOnTimeout: true })
+  }
+
+  /**
+   * Connects to a given `Multiaddr`. `addr` should include the id of the peer being
+   * dialed, it will be used for encryption verification.
+   *
+   * @async
+   * @param {Multiaddr} addr The address to dial
+   * @param {object} [options]
+   * @param {AbortSignal} [options.signal] An AbortController signal
+   * @returns {Promise<Connection>}
+   */
+  async connectToMultiaddr (addr, options = {}) {
+    addr = multiaddr(addr)
+    let conn
+    let controller
+
+    if (!options.signal) {
+      controller = new AbortController()
+      options.signal = controller.signal
+    }
+
+    try {
+      conn = await this.queue.add(() => this.transportManager.dial(addr, options))
+    } catch (err) {
+      if (err.name === 'TimeoutError') {
+        controller.abort()
+        err.code = codes.ERR_TIMEOUT
+      }
+      log.error('Error dialing address %s,', addr, err)
+      throw err
+    }
+
+    return conn
+  }
+
+  /**
+   * Connects to a given `PeerInfo` by dialing all of its known addresses.
+   * The dial to the first address that is successfully able to upgrade a connection
+   * will be used.
+   *
+   * @async
+   * @param {PeerInfo} peerInfo The remote peer to dial
+   * @param {object} [options]
+   * @param {AbortSignal} [options.signal] An AbortController signal
+   * @returns {Promise<Connection>}
+   */
+  async connectToPeer (peerInfo, options = {}) {
+    const addrs = peerInfo.multiaddrs.toArray()
+    for (const addr of addrs) {
+      try {
+        return await this.connectToMultiaddr(addr, options)
+      } catch (_) {
+        // The error is already logged, just move to the next addr
+        continue
+      }
+    }
+
+    const err = errCode(new Error('Could not dial peer, all addresses failed'), codes.ERR_CONNECTION_FAILED)
+    log.error(err)
+    throw err
+  }
+}
+
+module.exports = Dialer

src/errors.js

@@ -8,10 +8,16 @@ exports.messages = {
 exports.codes = {
   DHT_DISABLED: 'ERR_DHT_DISABLED',
   PUBSUB_NOT_STARTED: 'ERR_PUBSUB_NOT_STARTED',
+  ERR_CONNECTION_FAILED: 'ERR_CONNECTION_FAILED',
   ERR_NODE_NOT_STARTED: 'ERR_NODE_NOT_STARTED',
   ERR_NO_VALID_ADDRESSES: 'ERR_NO_VALID_ADDRESSES',
   ERR_DISCOVERED_SELF: 'ERR_DISCOVERED_SELF',
   ERR_DUPLICATE_TRANSPORT: 'ERR_DUPLICATE_TRANSPORT',
+  ERR_ENCRYPTION_FAILED: 'ERR_ENCRYPTION_FAILED',
   ERR_INVALID_KEY: 'ERR_INVALID_KEY',
-  ERR_TRANSPORT_UNAVAILABLE: 'ERR_TRANSPORT_UNAVAILABLE'
+  ERR_MUXER_UNAVAILABLE: 'ERR_MUXER_UNAVAILABLE',
+  ERR_TIMEOUT: 'ERR_TIMEOUT',
+  ERR_TRANSPORT_UNAVAILABLE: 'ERR_TRANSPORT_UNAVAILABLE',
+  ERR_TRANSPORT_DIAL_FAILED: 'ERR_TRANSPORT_DIAL_FAILED',
+  ERR_UNSUPPORTED_PROTOCOL: 'ERR_UNSUPPORTED_PROTOCOL'
 }