feat: ipfs.ping cli, http-api and core (#1342)

circle.yml

@@ -1,13 +1,10 @@
 # Warning: This file is automatically synced from https://github.com/ipfs/ci-sync so if you want to change it, please change it there and ask someone to sync all repositories.
 machine:
   node:
-    version: 8
+    version: 8.11.1
 
 test:
-  pre:
-    - npm run lint
   post:
-    - make test
     - npm run coverage -- --upload --providers coveralls
 
 dependencies:

package.json

@@ -72,9 +72,9 @@
     "execa": "~0.10.0",
     "expose-loader": "~0.7.5",
     "form-data": "^2.3.2",
-    "hat": "~0.0.3",
-    "interface-ipfs-core": "~0.65.7",
-    "ipfsd-ctl": "~0.34.0",
+    "hat": "0.0.3",
+    "interface-ipfs-core": "~0.66.2",
+    "ipfsd-ctl": "~0.37.0",
     "lodash": "^4.17.10",
     "mocha": "^5.1.1",
     "ncp": "^2.0.0",
@@ -105,7 +105,7 @@
     "hapi-set-header": "^1.0.2",
     "hoek": "^5.0.3",
     "human-to-milliseconds": "^1.0.0",
-    "ipfs-api": "^21.0.0",
+    "ipfs-api": "^22.0.0",
     "ipfs-bitswap": "~0.20.0",
     "ipfs-block": "~0.7.1",
     "ipfs-block-service": "~0.14.0",
@@ -148,7 +148,7 @@
     "multihashes": "~0.4.13",
     "once": "^1.4.0",
     "path-exists": "^3.0.0",
-    "peer-book": "~0.7.0",
+    "peer-book": "~0.8.0",
     "peer-id": "~0.10.7",
     "peer-info": "~0.14.1",
     "progress": "^2.0.0",

src/cli/commands/ping.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const pull = require('pull-stream')
+const print = require('../utils').print
+
+module.exports = {
+  command: 'ping <peerId>',
+
+  description: 'Measure the latency of a connection',
+
+  builder: {
+    count: {
+      alias: 'n',
+      type: 'integer',
+      default: 10
+    }
+  },
+
+  handler (argv) {
+    const peerId = argv.peerId
+    const count = argv.count || 10
+    pull(
+      argv.ipfs.pingPullStream(peerId, { count }),
+      pull.drain(({ success, time, text }) => {
+        // Check if it's a pong
+        if (success && !text) {
+          print(`Pong received: time=${time} ms`)
+        // Status response
+        } else {
+          print(text)
+        }
+      })
+    )
+  }
+}

src/core/components/index.js

@@ -16,6 +16,8 @@ exports.dag = require('./dag')
 exports.libp2p = require('./libp2p')
 exports.swarm = require('./swarm')
 exports.ping = require('./ping')
+exports.pingPullStream = require('./ping-pull-stream')
+exports.pingReadableStream = require('./ping-readable-stream')
 exports.files = require('./files')
 exports.bitswap = require('./bitswap')
 exports.pubsub = require('./pubsub')

src/core/components/ping-pull-stream.js

@@ -0,0 +1,104 @@
+'use strict'
+
+const debug = require('debug')
+const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
+const PeerId = require('peer-id')
+const pull = require('pull-stream')
+const Pushable = require('pull-pushable')
+const waterfall = require('async/waterfall')
+
+const log = debug('jsipfs:pingPullStream')
+log.error = debug('jsipfs:pingPullStream:error')
+
+module.exports = function pingPullStream (self) {
+  return (peerId, opts) => {
+    if (!self.isOnline()) {
+      return pull.error(new Error(OFFLINE_ERROR))
+    }
+
+    opts = Object.assign({ count: 10 }, opts)
+
+    const source = Pushable()
+
+    waterfall([
+      (cb) => getPeer(self._libp2pNode, source, peerId, cb),
+      (peer, cb) => runPing(self._libp2pNode, source, opts.count, peer, cb)
+    ], (err) => {
+      if (err) {
+        log.error(err)
+        source.push(getPacket({ success: false, text: err.toString() }))
+        source.end(err)
+      }
+    })
+
+    return source
+  }
+}
+
+function getPacket (msg) {
+  // Default msg
+  const basePacket = { success: true, time: 0, text: '' }
+  return Object.assign(basePacket, msg)
+}
+
+function getPeer (libp2pNode, statusStream, peerId, cb) {
+  let peer
+
+  try {
+    peer = libp2pNode.peerBook.get(peerId)
+  } catch (err) {
+    log('Peer not found in peer book, trying peer routing')
+    // Share lookup status just as in the go implemmentation
+    statusStream.push(getPacket({ text: `Looking up peer ${peerId}` }))
+
+    // Try to use peerRouting
+    try {
+      peerId = PeerId.createFromB58String(peerId)
+    } catch (err) {
+      return cb(Object.assign(err, {
+        message: `failed to parse peer address '${peerId}': input isn't valid multihash`
+      }))
+    }
+
+    return libp2pNode.peerRouting.findPeer(peerId, cb)
+  }
+
+  cb(null, peer)
+}
+
+function runPing (libp2pNode, statusStream, count, peer, cb) {
+  libp2pNode.ping(peer, (err, p) => {
+    if (err) {
+      return cb(err)
+    }
+
+    log('Got peer', peer)
+
+    let packetCount = 0
+    let totalTime = 0
+    statusStream.push(getPacket({ text: `PING ${peer.id.toB58String()}` }))
+
+    p.on('ping', (time) => {
+      statusStream.push(getPacket({ time: time }))
+      totalTime += time
+      packetCount++
+      if (packetCount >= count) {
+        const average = totalTime / count
+        p.stop()
+        statusStream.push(getPacket({ text: `Average latency: ${average}ms` }))
+        statusStream.end()
+      }
+    })
+
+    p.on('error', (err) => {
+      log.error(err)
+      p.stop()
+      statusStream.push(getPacket({ success: false, text: err.toString() }))
+      statusStream.end(err)
+    })
+
+    p.start()
+
+    return cb()
+  })
+}

src/core/components/ping-readable-stream.js

@@ -0,0 +1,7 @@
+'use strict'
+
+const toStream = require('pull-stream-to-stream')
+
+module.exports = function pingReadableStream (self) {
+  return (peerId, opts) => toStream.source(self.pingPullStream(peerId, opts))
+}

src/core/components/ping.js

@@ -1,9 +1,13 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const pull = require('pull-stream/pull')
 
 module.exports = function ping (self) {
-  return promisify((callback) => {
-    callback(new Error('Not implemented'))
+  return promisify((peerId, opts, cb) => {
+    pull(
+      self.pingPullStream(peerId, opts),
+      pull.collect(cb)
+    )
   })
 }

src/core/index.js

@@ -102,6 +102,8 @@ class IPFS extends EventEmitter {
     this.files = components.files(this)
     this.bitswap = components.bitswap(this)
     this.ping = components.ping(this)
+    this.pingPullStream = components.pingPullStream(this)
+    this.pingReadableStream = components.pingReadableStream(this)
     this.pubsub = components.pubsub(this)
     this.dht = components.dht(this)
     this.dns = components.dns(this)

src/http/api/resources/index.js

@@ -3,6 +3,7 @@
 exports.version = require('./version')
 exports.shutdown = require('./shutdown')
 exports.id = require('./id')
+exports.ping = require('./ping')
 exports.bootstrap = require('./bootstrap')
 exports.repo = require('./repo')
 exports.object = require('./object')

src/http/api/resources/ping.js

@@ -0,0 +1,48 @@
+'use strict'
+
+const Joi = require('joi')
+const pull = require('pull-stream')
+const toStream = require('pull-stream-to-stream')
+const ndjson = require('pull-ndjson')
+const PassThrough = require('readable-stream').PassThrough
+const pump = require('pump')
+
+exports.get = {
+  validate: {
+    query: Joi.object().keys({
+      n: Joi.alternatives()
+        .when('count', {
+          is: Joi.any().exist(),
+          then: Joi.any().forbidden(),
+          otherwise: Joi.number().integer().greater(0)
+        }),
+      count: Joi.number().integer().greater(0),
+      arg: Joi.string().required()
+    }).unknown()
+  },
+  handler: (request, reply) => {
+    const ipfs = request.server.app.ipfs
+    const peerId = request.query.arg
+    // Default count to 10
+    const count = request.query.n || request.query.count || 10
+
+    const source = pull(
+      ipfs.pingPullStream(peerId, { count: count }),
+      pull.map((chunk) => ({
+        Success: chunk.success,
+        Time: chunk.time,
+        Text: chunk.text
+      })),
+      ndjson.serialize()
+    )
+
+    // Streams from pull-stream-to-stream don't seem to be compatible
+    // with the stream2 readable interface
+    // see: https://github.com/hapijs/hapi/blob/c23070a3de1b328876d5e64e679a147fafb04b38/lib/response.js#L533
+    // and: https://github.com/pull-stream/pull-stream-to-stream/blob/e436acee18b71af8e71d1b5d32eee642351517c7/index.js#L28
+    const responseStream = toStream.source(source)
+    const stream2 = new PassThrough()
+    pump(responseStream, stream2)
+    return reply(stream2).type('application/json').header('X-Chunked-Output', '1')
+  }
+}

src/http/api/routes/index.js

@@ -9,6 +9,7 @@ module.exports = (server) => {
   require('./object')(server)
   require('./repo')(server)
   require('./config')(server)
+  require('./ping')(server)
   require('./swarm')(server)
   require('./bitswap')(server)
   require('./file')(server)

src/http/api/routes/ping.js

@@ -0,0 +1,16 @@
+'use strict'
+
+const resources = require('./../resources')
+
+module.exports = (server) => {
+  const api = server.select('API')
+
+  api.route({
+    method: '*',
+    path: '/api/v0/ping',
+    config: {
+      handler: resources.ping.get.handler,
+      validate: resources.ping.get.validate
+    }
+  })
+}

test/cli/commands.js

@@ -4,7 +4,8 @@
 const expect = require('chai').expect
 const runOnAndOff = require('../utils/on-and-off')
 
-const commandCount = 73
+const commandCount = 74
+
 describe('commands', () => runOnAndOff((thing) => {
   let ipfs