feat: first shot at ping implementaion

Author: JGAntunes

package.json

@@ -151,6 +151,7 @@
     "progress": "^2.0.0",
     "promisify-es6": "^1.0.3",
     "pull-abortable": "^4.1.1",
+    "pull-catch": "^1.0.0",
     "pull-defer": "^0.2.2",
     "pull-file": "^1.1.0",
     "pull-ndjson": "^0.1.1",

src/cli/commands/ping.js

@@ -1,5 +1,9 @@
 'use strict'
 
+const pull = require('pull-stream/pull')
+const drain = require('pull-stream/sinks/drain')
+const pullCatch = require('pull-catch')
+
 const print = require('../utils').print
 
 module.exports = {
@@ -19,26 +23,20 @@ module.exports = {
     const peerId = argv.peerId
     const count = argv.count || 10
 
-    print('PING ' + peerId)
-
-    let noOfTimes = 0
-    let totalTime = 0
-
-    const pingCb = (err, p) => {
-      if (err) {
-        throw err
-      }
-      let time = p.Time
-      totalTime = totalTime + time
-      noOfTimes = noOfTimes + 1
-      print('Pong received: time=' + time + ' ms')
-      if (noOfTimes === count) {
-        print('Average latency: ' + totalTime / count + 'ms')
-      }
-    }
-
-    for (let i = 0; i < count; i++) {
-      argv.ipfs.ping(peerId, pingCb)
-    }
+    pull(
+      argv.ipfs.pingPullStream(peerId, { count }),
+      pullCatch(err => {
+        throw err 
+      }),
+      drain(({ Time, Text }) => {
+        // Check if it's a pong
+        if (Time) {
+          print(`Pong received: time=${Time} ms`)
+        // Status response
+        } else {
+          print(Text)
+        }
+      })
+    )
   }
 }

src/core/components/ping.js

@@ -4,32 +4,51 @@ const promisify = require('promisify-es6')
 const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
 const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
-var Readable = require('stream').Readable
+const Readable = require('readable-stream').Readable
+
+function getPacket (msg) {
+  // Default msg
+  const basePacket = {Success: false, Time: 0, Text: ''}
+  // ndjson
+  return `${JSON.stringify(Object.assign({}, basePacket, msg))}\n`
+}
 
 module.exports = function ping (self) {
-  return promisify((peerId, cb) => {
+  return promisify((peerId, count, cb) => {
     if (!self.isOnline()) {
       return cb(new Error(OFFLINE_ERROR))
     }
 
-    var outputStream = new Readable()
-    outputStream._read = function (size) {
-    }
+    const source = new Readable({
+      read: function () {}
+    })
 
     let peer
     try {
       peer = self._libp2pNode.peerBook.get(peerId)
     } catch (err) {
+      // Conforming with go implemmentation, not sure if makes sense to log this
+      // since we perform no `findPeer`
+      source.push(getPacket({Success: true, Text: `Looking up peer ${peerId}`}))
       peer = new PeerInfo(PeerId.createFromB58String(peerId))
     }
 
     self._libp2pNode.ping(peer, (err, p) => {
-      p.once('ping', (time) => {
-        outputStream.push(JSON.stringify([{}, { Success: true, Time: time }, { Text: 'Average latency: ' + time + ' ms' }]))
-        outputStream.push(null)
-        p.stop()
-        cb(err, outputStream)
+      let packetCount = 0 
+      let totalTime = 0
+      source.push(getPacket({Success: true, Text: `PING ${peerId}`}))
+      p.on('ping', (time) => {
+        source.push(getPacket({ Success: true, Time: time }))
+        totalTime += time
+        packetCount++
+        if (packetCount >= count) {
+          const average = totalTime/count
+          p.stop()
+          source.push(getPacket({ Success: false, Text: `Average latency: ${average}ms`}))
+          source.push(null)
+        }
       })
     })
+    cb(null, source)
   })
 }

src/http/api/resources/ping.js

@@ -1,18 +1,36 @@
 'use strict'
 
+const Joi = require('joi')
 const boom = require('boom')
+const toStream = require('pull-stream-to-stream')
 
 exports = module.exports
 
-exports.get = (request, reply) => {
-  const ipfs = request.server.app.ipfs
-  const peerId = request.query.arg
+exports.get = {
+  validate: {
+    query: Joi.object().keys({
+      n: Joi.alternatives()
+        .when('count', {
+          is: true, then: Joi.any().forbidden(),
+          otherwise: Joi.number().greater(0)
+        }),
+      count: Joi.number().greater(0),
+      arg: Joi.string()
+    })
+  },
+  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
 
-  ipfs.ping(peerId, (err, outputStream) => {
-    if (err) {
-      return reply(boom.badRequest(err))
-    }
+    ipfs.ping(peerId, count, (err, sourceStream) => {
+      if (err) {
+        return reply(boom.badRequest(err))
+      }
+      console.log(sourceStream)
 
-    return reply(outputStream).type('application/json').header('x-chunked-output', '1')
-  })
+      return reply(sourceStream).type('application/json').header('x-chunked-output', '1')
+    })
+  }
 }

src/http/api/routes/ping.js

@@ -13,7 +13,8 @@ module.exports = (server) => {
         parse: false,
         output: 'stream'
       },
-      handler: resources.ping.get
+      handler: resources.ping.get.handler,
+      validate: resources.ping.get.validate
     }
   })
 }