feat: add introspection

package.json

@@ -49,6 +49,9 @@
     "sinon": "^9.0.0"
   },
   "dependencies": {
+    "@libp2p/observer-data": "^1.1.0",
+    "@libp2p/observer-proto": "^1.1.0",
+    "base32.js": "^0.1.0",
     "cids": "^1.1.5",
     "debug": "^4.3.1",
     "it-buffer": "^0.1.2",
@@ -60,7 +63,7 @@
     "libp2p-bootstrap": "^0.12.1",
     "libp2p-floodsub": "^0.24.1",
     "libp2p-gossipsub": "^0.8.0",
-    "libp2p-kad-dht": "^0.20.6",
+    "libp2p-kad-dht": "libp2p/js-libp2p-kad-dht#feat/peer-eviction",
     "libp2p-mplex": "^0.10.0",
     "libp2p-noise": "^2.0.0",
     "libp2p-secio": "^0.13.1",

src/cli/bin.js

@@ -42,12 +42,12 @@ const main = async (processArgs) => {
     .option('secio', {
       desc: 'Enables secio connection encryption',
       type: 'boolean',
-      default: true
+      default: false
     })
     .option('noise', {
       desc: 'Enables noise connection encryption',
       type: 'boolean',
-      default: false
+      default: true
     })
     .option('bootstrap', {
       alias: 'b',
@@ -63,7 +63,7 @@ const main = async (processArgs) => {
     .option('dht', {
       desc: 'Enables the DHT in full node mode',
       type: 'boolean',
-      default: false
+      default: true
     })
     .option('dhtClient', {
       desc: '(Not yet supported) Enables the DHT in client mode',
@@ -118,6 +118,10 @@ const main = async (processArgs) => {
   if (!argv.quiet) {
     // eslint-disable-next-line
     log('daemon has started')
+    log('daemon is listening on:')
+    daemon.libp2p.multiaddrs.forEach((ma) => {
+      log(`${ma}/p2p/${daemon.libp2p.peerId.toB58String()}`)
+    })
   }
 }
 

src/daemon.js

@@ -4,6 +4,7 @@
 
 const TCP = require('libp2p-tcp')
 const Libp2p = require('./libp2p')
+const createIntrospection = require('./introspection')
 const PeerId = require('peer-id')
 const ma = require('multiaddr')
 const CID = require('cids')
@@ -42,6 +43,8 @@ class Daemon {
   }) {
     this.multiaddr = ma(multiaddr)
     this.libp2p = libp2pNode
+    // TODO: Add custom port + enabled (y/n)
+    this.introspection = createIntrospection({ libp2p: this.libp2p })
     this.tcp = new TCP({ upgrader: passThroughUpgrader })
     this.listener = this.tcp.createListener((maConn) => {
       this.handleConnection(maConn)
@@ -165,8 +168,22 @@ class Daemon {
    */
   async start () {
     this._listen()
+    await this.introspection.start()
     await this.libp2p.start()
     await this.listener.listen(this.multiaddr)
+
+    this._start()
+  }
+
+  // TMP for demo
+  async _start () {
+    await new Promise(resolve => setTimeout(resolve, 20e3))
+    // Find providers
+    const cid = new CID('QmZKcfhEUbY6GumNmL7rga58qnRdMqMtv64smChiHJyfJf')
+    console.log('Searching for %s', cid.toString())
+    for await (const provider of this.libp2p.contentRouting.findProviders(cid, { maxNumProviders: 1 })) {
+      console.log('Found provider %j', provider)
+    }
   }
 
   /**
@@ -177,6 +194,7 @@ class Daemon {
    * @returns {Promise<void>}
    */
   async stop (options = { exit: false }) {
+    await this.introspection.stop()
     await this.libp2p.stop()
     await this.listener.close()
     if (options.exit) {

src/introspection/create-message.js

@@ -0,0 +1,71 @@
+'use strict'
+
+const {
+  createResponseServerMessage,
+  createRuntimeServerMessage,
+  createStateServerMessage
+} = require('./messages/server-message')
+
+const {
+  createConnection
+} = require('./messages/connections')
+
+const {
+  createState: createStateMessage
+} = require('./messages/states')
+
+const {
+  createDHT,
+  updateDHT,
+} = require('./messages/dht')
+
+const {
+  createCommandMessage
+} = require('./messages/command-response')
+
+const {
+  createBufferSegment
+} = require('./utils')
+
+// TODO: Perhaps use in this context
+function createVersion () {
+  const versionBuf = Buffer.alloc(4)
+  versionBuf.writeUInt32LE(1, 0)
+  return versionBuf
+}
+
+function createRuntimeMessage (options = {}, runtime = createRuntime(options)) {
+  const runtimePacket = createRuntimeServerMessage(runtime)
+  return createBufferSegment(runtimePacket)
+}
+
+function createCommandResponse (options = {}, response = createCommandMessage(options)) {
+  const responsePacket = createResponseServerMessage(response)
+  return createBufferSegment(responsePacket)
+}
+
+function createConnections (connections = []) {
+  const connectionsPbs = []
+
+  connections.forEach((c) => {
+    connectionsPbs.push(createConnection(c))
+  })
+
+  return connectionsPbs
+}
+
+function createState (connections, utcNow, dht, durationSnapshot) {
+  const state = createStateMessage(connections, utcNow, dht, durationSnapshot)
+  const statePacket = createStateServerMessage(state)
+  return createBufferSegment(statePacket)
+}
+
+module.exports = {
+  createDHT,
+  createConnections,
+  createRuntimeMessage,
+  createVersion,
+  createState,
+  createCommandResponse,
+  updateDHT
+}