feat: add pingReadableStream and pingPullStream

License: MIT
Signed-off-by: Alan Shaw <[email protected]>

Author: alanshaw

src/cli/commands/ping.js

@@ -22,13 +22,13 @@ module.exports = {
     const count = argv.count || 10
     pull(
       argv.ipfs.pingPullStream(peerId, { count }),
-      pull.drain(({ Time, Text }) => {
+      pull.drain(({ time, text }) => {
         // Check if it's a pong
-        if (Time) {
-          print(`Pong received: time=${Time} ms`)
+        if (time) {
+          print(`Pong received: time=${time} ms`)
         // Status response
         } else {
-          print(Text)
+          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,93 @@
+'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({ 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
+    return libp2pNode.peerRouting.findPeer(PeerId.createFromB58String(peerId), cb)
+  }
+  return 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,97 +1,13 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const debug = require('debug')
-const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
-const PeerId = require('peer-id')
 const pull = require('pull-stream/pull')
-const Pushable = require('pull-pushable')
-const ndjson = require('pull-ndjson')
-const waterfall = require('async/waterfall')
-
-const log = debug('jsipfs:ping')
-log.error = debug('jsipfs:ping:error')
 
 module.exports = function ping (self) {
-  return promisify((peerId, count, cb) => {
-    if (!self.isOnline()) {
-      return cb(new Error(OFFLINE_ERROR))
-    }
-
-    const source = Pushable()
-
-    const response = pull(
-      source,
-      ndjson.serialize()
+  return promisify((peerId, opts, cb) => {
+    pull(
+      self.pingPullStream(peerId, opts),
+      pull.collect(cb)
     )
-    waterfall([
-      getPeer.bind(null, self._libp2pNode, source, peerId),
-      runPing.bind(null, self._libp2pNode, source, count)
-    ], (err) => {
-      if (err) {
-        log.error(err)
-        source.push(getPacket({Text: err.toString()}))
-        source.end(err)
-      }
-    })
-
-    cb(null, response)
-  })
-}
-
-function getPacket (msg) {
-  // Default msg
-  const basePacket = {Success: false, 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({Success: true, Text: `Looking up peer ${peerId}`}))
-    // Try to use peerRouting
-    return libp2pNode.peerRouting.findPeer(PeerId.createFromB58String(peerId), cb)
-  }
-  return 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({Success: true, Text: `PING ${peer.id.toB58String()}`}))
-
-    p.on('ping', (time) => {
-      statusStream.push(getPacket({ Success: true, Time: time }))
-      totalTime += time
-      packetCount++
-      if (packetCount >= count) {
-        const average = totalTime / count
-        p.stop()
-        statusStream.push(getPacket({ Success: true, Text: `Average latency: ${average}ms` }))
-        statusStream.end()
-      }
-    })
-
-    p.on('error', (err) => {
-      log.error(err)
-      p.stop()
-      statusStream.push(getPacket({Text: err.toString()}))
-      statusStream.end(err)
-    })
-
-    p.start()
-
-    return 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/ping.js

@@ -1,13 +1,12 @@
 'use strict'
 
 const Joi = require('joi')
-const boom = require('boom')
+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 = module.exports
-
 exports.get = {
   validate: {
     query: Joi.object().keys({
@@ -26,18 +25,24 @@ exports.get = {
     const peerId = request.query.arg
     // Default count to 10
     const count = request.query.n || request.query.count || 10
-    ipfs.ping(peerId, count, (err, pullStream) => {
-      if (err) {
-        return reply(boom.badRequest(err))
-      }
-      // 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(pullStream)
-      const stream2 = new PassThrough()
-      pump(responseStream, stream2)
-      return reply(stream2).type('application/json').header('X-Chunked-Output', '1')
-    })
+
+    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')
   }
 }

test/core/ping.spec.js

@@ -32,7 +32,7 @@ function spawnNode ({ dht = false }, cb) {
   }, cb)
 }
 
-describe('ping', function () {
+describe.only('ping', function () {
   this.timeout(60 * 1000)
 
   if (!isNode) return
@@ -85,10 +85,10 @@ describe('ping', function () {
       const count = 3
       pull(
         ipfsdA.api.pingPullStream(ipfsdBId, { count }),
-        drain(({ Success, Time, Text }) => {
-          expect(Success).to.be.true()
+        drain(({ success, time, text }) => {
+          expect(success).to.be.true()
           // It's a pong
-          if (Time) {
+          if (time) {
             packetNum++
           }
         }, (err) => {
@@ -104,16 +104,16 @@ describe('ping', function () {
       const count = 2
       pull(
         ipfsdA.api.pingPullStream('unknown', { count }),
-        drain(({ Success, Time, Text }) => {
+        drain(({ success, time, text }) => {
           messageNum++
           // Assert that the ping command falls back to the peerRouting
           if (messageNum === 1) {
-            expect(Text).to.include('Looking up')
+            expect(text).to.include('Looking up')
           }
 
           // Fails accordingly while trying to use peerRouting
           if (messageNum === 2) {
-            expect(Success).to.be.false()
+            expect(success).to.be.false()
           }
         }, (err) => {
           expect(err).to.not.exist()
@@ -175,11 +175,9 @@ describe('ping', function () {
       parallel([
         ipfsdA.api.swarm.connect.bind(ipfsdA.api, bMultiaddr),
         ipfsdB.api.swarm.connect.bind(ipfsdB.api, cMultiaddr)
-      ], (err) => setTimeout(() => done(err), 500)) // FIXME timeout needed for connections to succeed  
+      ], (err) => setTimeout(() => done(err), 500)) // FIXME timeout needed for connections to succeed
     })
 
-
-
     after((done) => ipfsdA.stop(done))
     after((done) => ipfsdB.stop(done))
     after((done) => ipfsdC.stop(done))
@@ -190,15 +188,15 @@ describe('ping', function () {
       const count = 3
       pull(
         ipfsdA.api.pingPullStream(ipfsdCId, { count }),
-        drain(({ Success, Time, Text }) => {
+        drain(({ success, time, text }) => {
           messageNum++
-          expect(Success).to.be.true()
+          expect(success).to.be.true()
           // Assert that the ping command falls back to the peerRouting
           if (messageNum === 1) {
-            expect(Text).to.include('Looking up')
+            expect(text).to.include('Looking up')
           }
           // It's a pong
-          if (Time) {
+          if (time) {
             packetNum++
           }
         }, (err) => {