wip: more floodsub tests; CLI placeholder tests; initializer

package.json

@@ -50,7 +50,6 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs#readme",
   "devDependencies": {
-
     "aegir": "9.1.2",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
@@ -91,7 +90,7 @@
     "ipld-resolver": "^0.2.0",
     "isstream": "^0.1.2",
     "joi": "^9.2.0",
-    "libp2p-floodsub": "0.3.0",
+    "libp2p-floodsub": "0.3.1",
     "libp2p-ipfs": "^0.15.0",
     "libp2p-ipfs-browser": "^0.16.0",
     "lodash.flatmap": "^4.5.0",
@@ -149,4 +148,4 @@
     "nginnever <ginneversource@gmail.com>",
     "npmcdn-to-unpkg-bot <npmcdn-to-unpkg-bot@users.noreply.github.com>"
   ]
-}
+}

src/cli/commands/floodsub.js

@@ -0,0 +1,17 @@
+'use strict'
+
+// NOTE: Floodsub CLI is not tested. Tests will not be run until
+// https://github.com/ipfs/js-ipfs-api/pull/377
+// is merged
+module.exports = {
+  command: 'floodsub',
+
+  description: 'floodsub commands',
+
+  builder (yargs) {
+    return yargs
+      .commandDir('floodsub')
+  },
+
+  handler (argv) {}
+}

src/cli/commands/floodsub/pub.js

@@ -0,0 +1,28 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:floodsub')
+log.error = debug('cli:floodsub:error')
+
+module.exports = {
+  command: 'pub <topic> <message>',
+
+  describe: 'Publish a message to a topic',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.floodsub.pub(argv.topic, argv.message, (err) => {
+        if (err) {
+          throw err
+        }
+      })
+    })
+  }
+}

src/cli/commands/floodsub/sub.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:floodsub')
+log.error = debug('cli:floodsub:error')
+
+module.exports = {
+  command: 'sub <topic>',
+
+  describe: 'Subscribe to a topic',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.floodsub.sub(argv.topic, (err, stream) => {
+        if (err) {
+          throw err
+        }
+
+        console.log(stream.toString())
+      })
+    })
+  }
+}

src/core/components/floodsub.js

@@ -1,25 +1,19 @@
 'use strict'
 
-// const FloodSub = require('libp2p-floodsub')
-const FloodSub = require('./../../../node_modules/libp2p-floodsub/src')
+const FloodSub = require('libp2p-floodsub')
 const promisify = require('promisify-es6')
-const Stream = require('stream')
+const Readable = require('stream').Readable
 
 const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
 
 module.exports = function floodsub (self) {
   let fsub
 
-  return {
-    start: promisify(() => {
-      if (!self.isOnline()) {
-        throw OFFLINE_ERROR
-      }
-
-      fsub = new FloodSub(self._libp2pNode)
-      return self._libp2pNode
-    }),
+  function init () {
+    fsub = new FloodSub(self._libp2pNode)
+  }
 
+  return {
     sub: promisify((topic, options, callback) => {
       // TODO: Clarify with @diasdavid what to do with the `options.discover` param
       // Ref: https://github.com/ipfs/js-ipfs-api/pull/377/files#diff-f0c61c06fd5dc36b6f760b7ea97b1862R50
@@ -32,9 +26,11 @@ module.exports = function floodsub (self) {
         throw OFFLINE_ERROR
       }
 
-      let rs = new Stream()
-      rs.readable = true
-      rs._read = () => {}
+      if (!fsub) {
+        init()
+      }
+
+      let rs = new Readable()
       rs.cancel = () => fsub.unsubscribe(topic)
 
       fsub.on(topic, (data) => {
@@ -58,6 +54,10 @@ module.exports = function floodsub (self) {
         throw OFFLINE_ERROR
       }
 
+      if (!fsub) {
+        init()
+      }
+
       const buf = Buffer.isBuffer(data) ? data : new Buffer(data)
 
       try {

src/core/components/go-online.js

@@ -13,7 +13,6 @@ module.exports = function goOnline (self) {
         return cb(err)
       }
 
-      self.floodsub.start()
       self._bitswap = new Bitswap(
         self._libp2pNode.peerInfo,
         self._libp2pNode,

src/http-api/resources/floodsub.js

@@ -8,7 +8,7 @@ exports = module.exports
 
 exports.sub = {
   handler: (request, reply) => {
-    const discover = request.query.discover
+    const discover = request.query.discover || null
     const topic = request.params.topic
 
     request.server.app.ipfs.floodsub.sub(topic, { discover }, (err, stream) => {
@@ -20,6 +20,14 @@ exports.sub = {
         }).code(500)
       }
 
+      // hapi is not very clever and throws if no
+      // - _read method
+      // - _readableState object
+      // are there :(
+      if (!stream._read) {
+        stream._read = () => {}
+        stream._readableState = {}
+      }
       return reply(stream)
     })
   }
@@ -39,7 +47,7 @@ exports.pub = {
         }).code(500)
       }
 
-      return reply(true)
+      return reply()
     })
   }
 }

src/http-api/routes/floodsub.js

@@ -24,7 +24,7 @@ module.exports = (server) => {
 
   api.route({
     method: '*',
-    path: '/api/v0/floodsub/pub',
+    path: '/api/v0/pubsub/pub',
     config: {
       handler: resources.floodsub.pub.handler,
       validate: {

test/cli/test-commands.js

@@ -7,11 +7,17 @@ const ipfsBase = require('../utils/ipfs-exec')
 const ipfs = ipfsBase(repoPath)
 const describeOnlineAndOffline = require('../utils/on-and-off')
 
+// NOTE: Floodsub CLI tests will not be run until
+// https://github.com/ipfs/js-ipfs-api/pull/377
+// is merged. But we have the files ready to go
+// so the command count is bumped from 56 to 59
+const commandCount = 59
+
 describe('commands', () => {
   describeOnlineAndOffline(repoPath, () => {
     it('list the commands', () => {
       return ipfs('commands').then((out) => {
-        expect(out.split('\n')).to.have.length(56)
+        expect(out.split('\n')).to.have.length(commandCount)
       })
     })
   })
@@ -20,7 +26,7 @@ describe('commands', () => {
     return ipfsBase(repoPath, {
       cwd: '/tmp'
     })('commands').then((out) => {
-      expect(out.split('\n').length).to.equal(56)
+      expect(out.split('\n').length).to.equal(commandCount)
     })
   })
 })

test/cli/test-floodsub.js

@@ -0,0 +1,67 @@
+/* eslint max-nested-callbacks: ["error", 8] */
+/* eslint-env mocha */
+'use strict'
+
+const expect = require('chai').expect
+const HttpAPI = require('../../src/http-api')
+const createTempNode = require('../utils/temp-node')
+const repoPath = require('./index').repoPath
+const ipfs = require('../utils/ipfs-exec')(repoPath)
+
+// NOTE: Floodsub CLI tests will not be run until
+// https://github.com/ipfs/js-ipfs-api/pull/377
+// is merged
+describe.skip('floodsub', function () {
+  this.timeout(30 * 1000)
+  let node
+
+  const topic = 'nonscents'
+  const message = new Buffer('Some non cents.')
+
+  before((done) => {
+    createTempNode(1, (err, _node) => {
+      expect(err).to.not.exist
+      node = _node
+      node.goOnline((err) => {
+        expect(err).to.not.exist
+        done()
+      })
+    })
+  })
+
+  after((done) => {
+    node.goOffline(done)
+  })
+
+  describe('api running', () => {
+    let httpAPI
+    const called = true
+
+    before((done) => {
+      httpAPI = new HttpAPI(repoPath)
+      httpAPI.start((err) => {
+        expect(err).to.not.exist
+        done()
+      })
+    })
+
+    after((done) => {
+      httpAPI.stop((err) => {
+        expect(err).to.not.exist
+        done()
+      })
+    })
+
+    it('pub', () => {
+      return ipfs('floodsub', 'pub', topic, message).then((out) => {
+        expect(called).to.eql(true)
+      })
+    })
+
+    it('sub', () => {
+      return ipfs('floodsub', 'sub', topic).then((out) => {
+        expect(out).to.have.length.above(0)
+      })
+    })
+  })
+})

test/core/both/test-floodsub.js

@@ -34,16 +34,6 @@ describe('floodsub', () => {
   })
 
   describe('Floodsub API', () => {
-    describe('start', () => {
-      it('throws if offline', () => {
-        expect(() => nodeOffline.floodsub.start()).to.throw
-      })
-
-      it('success', () => {
-        expect(nodeA.floodsub.start()).to.exist
-      })
-    })
-
     describe('sub', () => {
       it('throws if offline', () => {
         expect(() => nodeOffline.floodsub.sub()).to.throw