feat: add delegated peer and content routing support (#242)

* feat: allow for configuring content and peer routing

* feat: support multiple peer and content routing modules

* docs: add delegated routing example

src/config.js

@@ -10,14 +10,16 @@ const OptionsSchema = Joi.object({
   peerInfo: Joi.object().required(),
   peerBook: Joi.object(),
   modules: Joi.object().keys({
-    transport: Joi.array().items(ModuleSchema).min(1).required(),
-    streamMuxer: Joi.array().items(ModuleSchema).allow(null),
     connEncryption: Joi.array().items(ModuleSchema).allow(null),
     connProtector: Joi.object().keys({
       protect: Joi.func().required()
     }).unknown(),
+    contentRouting: Joi.array().items(Joi.object()).allow(null),
+    dht: ModuleSchema.allow(null),
     peerDiscovery: Joi.array().items(ModuleSchema).allow(null),
-    dht: ModuleSchema.allow(null)
+    peerRouting: Joi.array().items(Joi.object()).allow(null),
+    streamMuxer: Joi.array().items(ModuleSchema).allow(null),
+    transport: Joi.array().items(ModuleSchema).min(1).required()
   }).required(),
   config: Joi.object().keys({
     peerDiscovery: Joi.object().allow(null),

src/content-routing.js

@@ -1,20 +1,82 @@
 'use strict'
 
+const tryEach = require('async/tryEach')
+const parallel = require('async/parallel')
+const errCode = require('err-code')
+
 module.exports = (node) => {
+  const routers = node._modules.contentRouting || []
+
+  // If we have the dht, make it first
+  if (node._dht) {
+    routers.unshift(node._dht)
+  }
+
   return {
-    findProviders: (key, timeout, callback) => {
-      if (!node._dht) {
-        return callback(new Error('DHT is not available'))
+    /**
+     * Iterates over all content routers in series to find providers of the given key.
+     * Once a content router succeeds, iteration will stop.
+     *
+     * @param {CID} key The CID key of the content to find
+     * @param {object} options
+     * @param {number} options.maxTimeout How long the query should run
+     * @param {function(Error, Result<Array>)} callback
+     * @returns {void}
+     */
+    findProviders: (key, options, callback) => {
+      if (!routers.length) {
+        return callback(errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE'))
       }
 
-      node._dht.findProviders(key, timeout, callback)
+      if (typeof options === 'function') {
+        callback = options
+        options = {}
+      } else if (typeof options === 'number') { // This can be deprecated in a future release
+        options = {
+          maxTimeout: options
+        }
+      }
+
+      const tasks = routers.map((router) => {
+        return (cb) => router.findProviders(key, options, (err, results) => {
+          if (err) {
+            return cb(err)
+          }
+
+          // If we don't have any results, we need to provide an error to keep trying
+          if (!results || Object.keys(results).length === 0) {
+            return cb(errCode(new Error('not found'), 'NOT_FOUND'), null)
+          }
+
+          cb(null, results)
+        })
+      })
+
+      tryEach(tasks, (err, results) => {
+        if (err && err.code !== 'NOT_FOUND') {
+          return callback(err)
+        }
+        results = results || []
+        callback(null, results)
+      })
     },
+
+    /**
+     * Iterates over all content routers in parallel to notify it is
+     * a provider of the given key.
+     *
+     * @param {CID} key The CID key of the content to find
+     * @param {function(Error)} callback
+     * @returns {void}
+     */
     provide: (key, callback) => {
-      if (!node._dht) {
-        return callback(new Error('DHT is not available'))
+      if (!routers.length) {
+        return callback(errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE'))
       }
 
-      node._dht.provide(key, callback)
+      parallel(routers.map((router) => {
+        return (cb) => router.provide(key, cb)
+      }), callback)
     }
   }
 }

src/index.js

@@ -20,8 +20,6 @@ const pubsub = require('./pubsub')
 const getPeerInfo = require('./get-peer-info')
 const validateConfig = require('./config').validate
 
-exports = module.exports
-
 const NOT_STARTED_ERROR_MESSAGE = 'The libp2p node is not started yet'
 
 class Node extends EventEmitter {
@@ -102,6 +100,7 @@ class Node extends EventEmitter {
     }
 
     // Attach remaining APIs
+    // peer and content routing will automatically get modules from _modules and _dht
     this.peerRouting = peerRouting(this)
     this.contentRouting = contentRouting(this)
     this.dht = dht(this)

src/peer-routing.js

@@ -1,13 +1,58 @@
 'use strict'
 
+const tryEach = require('async/tryEach')
+const errCode = require('err-code')
+
 module.exports = (node) => {
+  const routers = node._modules.peerRouting || []
+
+  // If we have the dht, make it first
+  if (node._dht) {
+    routers.unshift(node._dht)
+  }
+
   return {
-    findPeer: (id, callback) => {
-      if (!node._dht) {
-        return callback(new Error('DHT is not available'))
+    /**
+     * Iterates over all peer routers in series to find the given peer.
+     *
+     * @param {String} id The id of the peer to find
+     * @param {object} options
+     * @param {number} options.maxTimeout How long the query should run
+     * @param {function(Error, Result<Array>)} callback
+     * @returns {void}
+     */
+    findPeer: (id, options, callback) => {
+      if (!routers.length) {
+        callback(errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE'))
       }
 
-      node._dht.findPeer(id, callback)
+      if (typeof options === 'function') {
+        callback = options
+        options = {}
+      }
+
+      const tasks = routers.map((router) => {
+        return (cb) => router.findPeer(id, options, (err, result) => {
+          if (err) {
+            return cb(err)
+          }
+
+          // If we don't have a result, we need to provide an error to keep trying
+          if (!result || Object.keys(result).length === 0) {
+            return cb(errCode(new Error('not found'), 'NOT_FOUND'), null)
+          }
+
+          cb(null, result)
+        })
+      })
+
+      tryEach(tasks, (err, results) => {
+        if (err && err.code !== 'NOT_FOUND') {
+          return callback(err)
+        }
+        results = results || []
+        callback(null, results)
+      })
     }
   }
 }