feat: use PubSub API directly from libp2p

.travis.yml

@@ -3,15 +3,11 @@ language: node_js
 
 matrix:
   include:
-  - node_js: 6
-    env: CXX=g++-4.8
   - node_js: 8
     env: CXX=g++-4.8
 
 script:
-  - npm run lint
   - npm run test
-  - make test
 
 before_script:
   - export DISPLAY=:99.0

circle.yml

@@ -4,7 +4,10 @@ machine:
     version: stable
 
 test:
+  pre:
+    - npm run lint
   post:
+    - make test
     - npm run coverage -- --upload --providers coveralls
 
 dependencies:

package.json

@@ -119,7 +119,7 @@
     "is-ipfs": "^0.3.2",
     "is-stream": "^1.1.0",
     "joi": "^13.1.2",
-    "libp2p": "~0.16.5",
+    "libp2p": "~0.17.0",
     "libp2p-circuit": "~0.1.4",
     "libp2p-floodsub": "~0.14.1",
     "libp2p-kad-dht": "~0.8.0",

src/core/components/bootstrap.js

@@ -1,9 +1,24 @@
 'use strict'
 
 const defaultNodes = require('../runtime/config-nodejs.json').Bootstrap
-const MultiAddr = require('multiaddr')
+const Multiaddr = require('multiaddr')
 const promisify = require('promisify-es6')
 
+function isValid (ma) {
+  if (typeof ma === 'string') {
+    try {
+      ma = new Multiaddr(ma)
+      return Boolean(ma)
+    } catch (err) {
+      return false
+    }
+  } else if (ma) {
+    return Multiaddr.isMultiaddr(ma)
+  } else {
+    return false
+  }
+}
+
 module.exports = function bootstrap (self) {
   return {
     list: promisify((callback) => {
@@ -17,15 +32,13 @@ module.exports = function bootstrap (self) {
     add: promisify((multiaddr, args, callback) => {
       if (typeof args === 'function') {
         callback = args
-        args = {default: false}
-      }
-      try {
-        if (multiaddr) 
-          new MultiAddr(multiaddr)
+        args = { default: false }
       }
-      catch (err) {
-        return setImmediate(() => callback(err))
+
+      if (multiaddr && !isValid(multiaddr)) {
+        return setImmediate(() => callback(new Error('Not valid multiaddr')))
       }
+
       self._repo.config.get((err, config) => {
         if (err) {
           return callback(err)
@@ -51,13 +64,10 @@ module.exports = function bootstrap (self) {
         callback = args
         args = {all: false}
       }
-      try {
-        if (multiaddr) 
-          new MultiAddr(multiaddr)
-      }
-      catch (err) {
-        return setImmediate(() => callback(err))
+      if (multiaddr && !isValid(multiaddr)) {
+        return setImmediate(() => callback(new Error('Not valid multiaddr')))
       }
+
       self._repo.config.get((err, config) => {
         if (err) {
           return callback(err)

src/core/components/libp2p.js

@@ -21,6 +21,7 @@ module.exports = function libp2p (self) {
           bootstrap: get(config, 'Bootstrap'),
           modules: self._libp2pModules,
           // EXPERIMENTAL
+          pubsub: get(self._options, 'EXPERIMENTAL.pubsub', false),
           dht: get(self._options, 'EXPERIMENTAL.dht', false),
           relay: {
             enabled: get(config, 'EXPERIMENTAL.relay.enabled', false),
@@ -50,9 +51,7 @@ module.exports = function libp2p (self) {
         })
 
         self._libp2pNode.start((err) => {
-          if (err) {
-            return callback(err)
-          }
+          if (err) { return callback(err) }
 
           self._libp2pNode.peerInfo.multiaddrs.forEach((ma) => {
             console.log('Swarm listening on', ma.toString())

src/core/components/pubsub.js

@@ -1,96 +1,48 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const setImmediate = require('async/setImmediate')
-
-const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
 
 module.exports = function pubsub (self) {
   return {
     subscribe: (topic, options, handler, callback) => {
-      if (!self.isOnline()) {
-        throw new Error(OFFLINE_ERROR)
-      }
-
       if (typeof options === 'function') {
         callback = handler
         handler = options
         options = {}
       }
 
-      function subscribe (cb) {
-        if (self._pubsub.listenerCount(topic) === 0) {
-          self._pubsub.subscribe(topic)
-        }
-
-        self._pubsub.on(topic, handler)
-        setImmediate(cb)
-      }
-
       if (!callback) {
         return new Promise((resolve, reject) => {
-          subscribe((err) => {
+          self._libp2pNode.pubsub.subscribe(topic, options, handler, (err) => {
             if (err) {
               return reject(err)
             }
             resolve()
           })
         })
       } else {
-        subscribe(callback)
+        self._libp2pNode.pubsub.subscribe(topic, options, handler, callback)
       }
     },
 
     unsubscribe: (topic, handler) => {
-      self._pubsub.removeListener(topic, handler)
-
-      if (self._pubsub.listenerCount(topic) === 0) {
-        self._pubsub.unsubscribe(topic)
-      }
+      self._libp2pNode.pubsub.unsubscribe(topic, handler)
     },
 
     publish: promisify((topic, data, callback) => {
-      if (!self.isOnline()) {
-        return setImmediate(() => callback(new Error(OFFLINE_ERROR)))
-      }
-
-      if (!Buffer.isBuffer(data)) {
-        return setImmediate(() => callback(new Error('data must be a Buffer')))
-      }
-
-      self._pubsub.publish(topic, data)
-      setImmediate(() => callback())
+      self._libp2pNode.pubsub.publish(topic, data, callback)
     }),
 
     ls: promisify((callback) => {
-      if (!self.isOnline()) {
-        return setImmediate(() => callback(new Error(OFFLINE_ERROR)))
-      }
-
-      const subscriptions = Array.from(self._pubsub.subscriptions)
-
-      setImmediate(() => callback(null, subscriptions))
+      self._libp2pNode.pubsub.ls(callback)
     }),
 
     peers: promisify((topic, callback) => {
-      if (!self.isOnline()) {
-        return setImmediate(() => callback(new Error(OFFLINE_ERROR)))
-      }
-
-      if (typeof topic === 'function') {
-        callback = topic
-        topic = null
-      }
-
-      const peers = Array.from(self._pubsub.peers.values())
-        .filter((peer) => topic ? peer.topics.has(topic) : true)
-        .map((peer) => peer.info.id.toB58String())
-
-      setImmediate(() => callback(null, peers))
+      self._libp2pNode.pubsub.peers(topic, callback)
     }),
 
     setMaxListeners (n) {
-      return self._pubsub.setMaxListeners(n)
+      self._libp2pNode.pubsub.setMaxListeners(n)
     }
   }
 }

src/core/components/start.js

@@ -2,8 +2,6 @@
 
 const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
-const FloodSub = require('libp2p-floodsub')
-const NoFloodSub = require('./no-floodsub')
 const setImmediate = require('async/setImmediate')
 const promisify = require('promisify-es6')
 
@@ -38,9 +36,7 @@ module.exports = (self) => {
       (cb) => self.preStart(cb),
       (cb) => self.libp2p.start(cb)
     ], (err) => {
-      if (err) {
-        return done(err)
-      }
+      if (err) { return done(err) }
 
       self._bitswap = new Bitswap(
         self._libp2pNode,
@@ -50,11 +46,7 @@ module.exports = (self) => {
 
       self._bitswap.start()
       self._blockService.setExchange(self._bitswap)
-
-      self._pubsub = self._options.EXPERIMENTAL.pubsub
-        ? new FloodSub(self._libp2pNode)
-        : new NoFloodSub()
-      self._pubsub.start(done)
+      done()
     })
   })
 }

src/core/components/stop.js

@@ -32,7 +32,6 @@ module.exports = (self) => {
     self._bitswap.stop()
 
     series([
-      (cb) => self._pubsub.stop(cb),
       (cb) => self.libp2p.stop(cb),
       (cb) => self._repo.close(cb)
     ], done)

test/core/bitswap.spec.js

@@ -16,10 +16,9 @@ const isNode = require('detect-node')
 const multihashing = require('multihashing-async')
 const CID = require('cids')
 
-const DaemonFactory = require('ipfsd-ctl')
-const df = DaemonFactory.create({ type: 'js' })
-
-const dfProc = DaemonFactory.create({ type: 'proc' })
+const IPFSFactory = require('ipfsd-ctl')
+const fDaemon = IPFSFactory.create({ type: 'js' })
+const fInProc = IPFSFactory.create({ type: 'proc' })
 
 // This gets replaced by '../utils/create-repo-browser.js' in the browser
 const createTempRepo = require('../utils/create-repo-nodejs.js')
@@ -69,7 +68,7 @@ function connectNodes (remoteNode, inProcNode, callback) {
 let nodes = []
 
 function addNode (inProcNode, callback) {
-  df.spawn({
+  fDaemon.spawn({
     exec: './src/cli/bin.js',
     config: {
       Addresses: {
@@ -119,7 +118,7 @@ describe('bitswap', function () {
       })
     }
 
-    dfProc.spawn({ exec: IPFS, config }, (err, _ipfsd) => {
+    fInProc.spawn({ exec: IPFS, config: config }, (err, _ipfsd) => {
       expect(err).to.not.exist()
       nodes.push(_ipfsd)
       inProcNode = _ipfsd.api