chore: release version v0.19.0

.travis.yml

@@ -4,25 +4,16 @@ language: node_js
 
 matrix:
   include:
-  - node_js: 6
+  - node_js: 'stable'
     env: CXX=g++-4.8
-  - node_js: 8
-    env: CXX=g++-4.8
-    #  - node_js: stable
-    #    env: CXX=g++-4.8
 
 script:
-  - npm run lint
   - npm run test
-  - npm run coverage
 
 before_script:
   - export DISPLAY=:99.0
   - sh -e /etc/init.d/xvfb start
 
-after_success:
-  - npm run coverage-publish
-
 addons:
   firefox: 'latest'
   apt:

CHANGELOG.md

@@ -1,3 +1,43 @@
+<a name="0.19.0"></a>
+# [0.19.0](https://github.com/libp2p/js-libp2p/compare/v0.18.0...v0.19.0) (2018-03-15)
+
+
+
+<a name="0.18.0"></a>
+# [0.18.0](https://github.com/libp2p/js-libp2p/compare/v0.17.0...v0.18.0) (2018-02-19)
+
+
+
+<a name="0.17.0"></a>
+# [0.17.0](https://github.com/libp2p/js-libp2p/compare/v0.16.5...v0.17.0) (2018-02-16)
+
+
+### Bug Fixes
+
+* use correct reference to floodSub ([947eaf1](https://github.com/libp2p/js-libp2p/commit/947eaf1))
+
+
+### Features
+
+* add pubsub to libp2p ([0c543b7](https://github.com/libp2p/js-libp2p/commit/0c543b7))
+
+
+
+<a name="0.16.5"></a>
+## [0.16.5](https://github.com/libp2p/js-libp2p/compare/v0.16.4...v0.16.5) (2018-02-14)
+
+
+
+<a name="0.16.4"></a>
+## [0.16.4](https://github.com/libp2p/js-libp2p/compare/v0.16.3...v0.16.4) (2018-02-09)
+
+
+
+<a name="0.16.3"></a>
+## [0.16.3](https://github.com/libp2p/js-libp2p/compare/v0.16.2...v0.16.3) (2018-02-08)
+
+
+
 <a name="0.16.2"></a>
 ## [0.16.2](https://github.com/libp2p/js-libp2p/compare/v0.16.1...v0.16.2) (2018-02-07)
 

README.md

@@ -249,9 +249,13 @@ class Node extends libp2p {
 
 > PeerInfo instance of the node
 
+#### `libp2p.pubsub`
+
+> Same API as IPFS PubSub, defined in the [CORE API Spec](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/PUBSUB.md). Just replace `ipfs` by `libp2p` and you are golden.
+
 ---------------------
 
-`DHT methods exposed`
+`DHT methods also exposed for the time being`
 
 #### `libp2p.dht.put(key, value, callback)`
 
@@ -273,6 +277,113 @@ class Node extends libp2p {
 [multiaddr]: https://github.com/multiformats/js-multiaddr
 [Connection]: https://github.com/libp2p/interface-connection
 
+-------
+
+### Switch Stats API
+
+##### `libp2p.stats.emit('update')`
+
+Every time any stat value changes, this object emits an `update` event.
+
+#### Global stats
+
+##### `libp2p.stats.global.snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+- dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+- dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+##### `libp2p.stats.global.movingAverages`
+
+Returns an object containing the following keys:
+
+- dataSent
+- dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-transport stats
+
+##### `libp2p.stats.transports()`
+
+Returns an array containing the tags (string) for each observed transport.
+
+##### `libp2p.stats.forTransport(transportTag).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+- dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+- dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+##### `libp2p.stats.forTransport(transportTag).movingAverages`
+
+Returns an object containing the following keys:
+
+ dataSent
+ dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-protocol stats
+
+##### `libp2p.stats.protocols()`
+
+Returns an array containing the tags (string) for each observed protocol.
+
+##### `libp2p.stats.forProtocol(protocolTag).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+- dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+- dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+
+##### `libp2p.stats.forProtocol(protocolTag).movingAverages`
+
+Returns an object containing the following keys:
+
+- dataSent
+- dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Per-peer stats
+
+##### `libp2p.stats.peers()`
+
+Returns an array containing the peerIDs (B58-encoded string) for each observed peer.
+
+##### `libp2p.stats.forPeer(peerId:String).snapshot`
+
+Should return a stats snapshot, which is an object containing the following keys and respective values:
+
+- dataSent: amount of bytes sent, [Big](https://github.com/MikeMcl/big.js#readme) number
+- dataReceived: amount of bytes received, [Big](https://github.com/MikeMcl/big.js#readme) number
+
+
+##### `libp2p.stats.forPeer(peerId:String).movingAverages`
+
+Returns an object containing the following keys:
+
+- dataSent
+- dataReceived
+
+Each one of them contains an object that has a key for each interval (`60000`, `300000` and `900000` miliseconds).
+
+Each one of these values is [an exponential moving-average instance](https://github.com/pgte/moving-average#readme).
+
+#### Stats update interval
+
+Stats are not updated in real-time. Instead, measurements are buffered and stats are updated at an interval. The maximum interval can be defined through the `Switch` constructor option `stats.computeThrottleTimeout`, defined in miliseconds.
+
+
 ## Development
 
 **Clone and install dependencies:**

circle.yml

@@ -2,6 +2,12 @@ machine:
   node:
     version: stable
 
+test:
+  pre:
+    - npm run lint
+  post:
+    - npm run coverage -- --upload --providers coveralls
+
 dependencies:
   pre:
     - google-chrome --version
@@ -15,4 +21,4 @@ dependencies:
     - sudo apt-get install -f
     - sudo apt-get install --only-upgrade lsb-base
     - sudo dpkg -i google-chrome.deb
-    - google-chrome --version
+    - google-chrome --version

examples/chat/src/libp2p-bundle.js

@@ -6,7 +6,7 @@ const WS = require('libp2p-websockets')
 const Railing = require('libp2p-railing')
 const spdy = require('libp2p-spdy')
 const KadDHT = require('libp2p-kad-dht')
-const multiplex = require('libp2p-multiplex')
+const mplex = require('libp2p-mplex')
 const secio = require('libp2p-secio')
 const libp2p = require('../../..')
 
@@ -17,7 +17,7 @@ function mapMuxers (list) {
     }
     switch (pref.trim().toLowerCase()) {
       case 'spdy': return spdy
-      case 'multiplex': return multiplex
+      case 'mplex': return mplex
       default:
         throw new Error(pref + ' muxer not available')
     }
@@ -31,7 +31,7 @@ function getMuxers (muxers) {
   } else if (muxers) {
     return mapMuxers(muxers)
   } else {
-    return [multiplex, spdy]
+    return [mplex, spdy]
   }
 }
 

examples/chat/src/listener.js

@@ -21,7 +21,7 @@ PeerId.createFromJSON(require('./peer-id-listener'), (err, idListener) => {
       throw err
     }
 
-    nodeListener.swarm.on('peer-mux-established', (peerInfo) => {
+    nodeListener.switch.on('peer-mux-established', (peerInfo) => {
       console.log(peerInfo.id.toB58String())
     })
 

examples/discovery-mechanisms/1.js

@@ -2,7 +2,7 @@
 
 const libp2p = require('libp2p')
 const TCP = require('libp2p-tcp')
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const PeerInfo = require('peer-info')
 const Railing = require('libp2p-railing')
@@ -26,7 +26,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       discovery: [new Railing(bootstrapers)]

examples/discovery-mechanisms/2.js

@@ -2,7 +2,7 @@
 
 const libp2p = require('libp2p')
 const TCP = require('libp2p-tcp')
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const PeerInfo = require('peer-info')
 const MulticastDNS = require('libp2p-mdns')
@@ -14,7 +14,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       discovery: [new MulticastDNS(peerInfo, { interval: 1000 })]

examples/discovery-mechanisms/README.md

@@ -8,7 +8,7 @@ These mechanisms save configuration and enable a node to operate without any exp
 
 ## 1. Bootstrap list of Peers when booting a node
 
-For this demo, we will connect to IPFS default bootstrapper nodes and so, we will need to support the same set of features those nodes have, that are: TCP, multiplex and SECIO. You can see the complete example at [1.js](./1.js).
+For this demo, we will connect to IPFS default bootstrapper nodes and so, we will need to support the same set of features those nodes have, that are: TCP, mplex and SECIO. You can see the complete example at [1.js](./1.js).
 
 First, we create our libp2p bundle.
 
@@ -18,7 +18,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       discovery: [new Railing(bootstrapers)]
@@ -111,7 +111,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       // We set the interval here to 1 second so that is faster to observe. The

examples/echo/src/libp2p-bundle.js

@@ -6,7 +6,7 @@ const WS = require('libp2p-websockets')
 const Railing = require('libp2p-railing')
 const spdy = require('libp2p-spdy')
 const KadDHT = require('libp2p-kad-dht')
-const multiplex = require('libp2p-multiplex')
+const mplex = require('libp2p-mplex')
 const secio = require('libp2p-secio')
 const libp2p = require('../../..')
 
@@ -17,7 +17,7 @@ function mapMuxers (list) {
     }
     switch (pref.trim().toLowerCase()) {
       case 'spdy': return spdy
-      case 'multiplex': return multiplex
+      case 'mplex': return mplex
       default:
         throw new Error(pref + ' muxer not available')
     }
@@ -31,7 +31,7 @@ function getMuxers (muxers) {
   } else if (muxers) {
     return mapMuxers(muxers)
   } else {
-    return [multiplex, spdy]
+    return [mplex, spdy]
   }
 }
 

examples/echo/src/listener.js

@@ -27,7 +27,7 @@ series([
     listenerPeerInfo.multiaddrs.add('/ip4/0.0.0.0/tcp/10333')
     listenerNode = new Node(listenerPeerInfo)
 
-    listenerNode.swarm.on('peer-mux-established', (peerInfo) => {
+    listenerNode.switch.on('peer-mux-established', (peerInfo) => {
       console.log('received dial to me from:', peerInfo.id.toB58String())
     })
 

examples/libp2p-in-the-browser/1/package.json

@@ -18,7 +18,7 @@
   "dependencies": {
     "detect-dom-ready": "^1.0.2",
     "libp2p": "^0.13.0",
-    "libp2p-multiplex": "^0.5.0",
+    "libp2p-mplex": "^0.6.0",
     "libp2p-railing": "^0.7.1",
     "libp2p-secio": "^0.8.1",
     "libp2p-spdy": "^0.11.0",

examples/libp2p-in-the-browser/1/src/browser-bundle.js

@@ -3,7 +3,7 @@
 const WebRTCStar = require('libp2p-webrtc-star')
 const WebSockets = require('libp2p-websockets')
 
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SPDY = require('libp2p-spdy')
 const SECIO = require('libp2p-secio')
 
@@ -37,7 +37,7 @@ class Node extends libp2p {
       ],
       connection: {
         muxer: [
-          Multiplex,
+          Mplex,
           SPDY
         ],
         crypto: [SECIO]

examples/peer-and-content-routing/1.js

@@ -2,7 +2,7 @@
 
 const libp2p = require('libp2p')
 const TCP = require('libp2p-tcp')
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const PeerInfo = require('peer-info')
 const KadDHT = require('libp2p-kad-dht')
@@ -15,7 +15,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       // we add the DHT module that will enable Peer and Content Routing

examples/peer-and-content-routing/2.js

@@ -2,7 +2,7 @@
 
 const libp2p = require('libp2p')
 const TCP = require('libp2p-tcp')
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const PeerInfo = require('peer-info')
 const CID = require('cids')
@@ -16,7 +16,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       // we add the DHT module that will enable Peer and Content Routing

examples/peer-and-content-routing/README.md

@@ -18,7 +18,7 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
       // we add the DHT module that will enable Peer and Content Routing

examples/protocol-and-stream-muxing/README.md

@@ -88,7 +88,7 @@ The example above would require a node to create a whole new connection for ever
 
 Stream multiplexing is a old concept, in fact it happens in many of the layers of the [OSI System](https://en.wikipedia.org/wiki/OSI_model), in libp2p we make this feature to our avail by letting the user pick which module for stream multiplexing to use.
 
-Currently, we have two available [libp2p-spdy](https://github.com/libp2p/js-libp2p-spdy) and [libp2p-multiplex](https://github.com/libp2p/js-libp2p-multiplex) and pluging them in is as easy as adding another transport. Let's revisit our libp2p bundle.
+Currently, we have two available [libp2p-spdy](https://github.com/libp2p/js-libp2p-spdy) and [libp2p-mplex](https://github.com/libp2p/js-libp2p-mplex) and pluging them in is as easy as adding another transport. Let's revisit our libp2p bundle.
 
 ```JavaScript
 const SPDY = require('libp2p-spdy')

examples/pubsub/1.js

@@ -2,11 +2,10 @@
 
 const libp2p = require('libp2p')
 const TCP = require('libp2p-tcp')
-const Multiplex = require('libp2p-multiplex')
+const Mplex = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const PeerInfo = require('peer-info')
 const MulticastDNS = require('libp2p-mdns')
-const FloodSub = require('libp2p-floodsub')
 const waterfall = require('async/waterfall')
 const parallel = require('async/parallel')
 const series = require('async/series')
@@ -16,10 +15,12 @@ class MyBundle extends libp2p {
     const modules = {
       transport: [new TCP()],
       connection: {
-        muxer: [Multiplex],
+        muxer: [Mplex],
         crypto: [SECIO]
       },
-      discovery: [new MulticastDNS(peerInfo, { interval: 2000 })]
+      discovery: [
+        new MulticastDNS(peerInfo, { interval: 2000 })
+      ]
     }
     super(modules, peerInfo)
   }
@@ -47,22 +48,25 @@ parallel([
   const node1 = nodes[0]
   const node2 = nodes[1]
 
-  const fs1 = new FloodSub(node1)
-  const fs2 = new FloodSub(node2)
-
   series([
-    (cb) => fs1.start(cb),
-    (cb) => fs2.start(cb),
     (cb) => node1.once('peer:discovery', (peer) => node1.dial(peer, cb)),
     (cb) => setTimeout(cb, 500)
   ], (err) => {
     if (err) { throw err }
 
-    fs2.on('news', (msg) => console.log(msg.from, msg.data.toString()))
-    fs2.subscribe('news')
-
-    setInterval(() => {
-      fs1.publish('news', Buffer.from('Bird bird bird, bird is the word!'))
-    }, 1000)
+    // Subscribe to the topic 'news'
+    node1.pubsub.subscribe('news',
+      (msg) => console.log(msg.from, msg.data.toString()),
+      () => {
+        setInterval(() => {
+          // Publish the message on topic 'news'
+          node2.pubsub.publish(
+            'news',
+            Buffer.from('Bird bird bird, bird is the word!'),
+            () => {}
+          )
+        }, 1000)
+      }
+    )
   })
 })

examples/pubsub/README.md

@@ -1,8 +1,6 @@
 # Publish Subscribe
 
-Publish Subscribe is something out of scope for the modular networking stack that is libp2p, however, it is something that is enabled through the primitives that libp2p offers and so it has become one of the most interesting use cases for libp2p.
-
-Currently, we have a PubSub implementation, [libp2p-floodsub](https://github.com/libp2p/js-libp2p-floodsub) and many more being researched at [research-pubsub](https://github.com/libp2p/research-pubsub).
+Publish Subscribe is also included on the stack. Currently, we have on PubSub implementation which we ship by default [libp2p-floodsub](https://github.com/libp2p/js-libp2p-floodsub), with many more being researched at [research-pubsub](https://github.com/libp2p/research-pubsub).
 
 We've seen many interesting use cases appear with this, here are some highlights:
 
@@ -12,30 +10,31 @@ We've seen many interesting use cases appear with this, here are some highlights
 
 ## 1. Setting up a simple PubSub network on top of libp2p
 
-For this example, we will use MulticastDNS for automatic Peer Discovery and libp2p-floodsub to give us the PubSub primitives we are looking for. This example is based the previous examples found in [Discovery Mechanisms](../discovery-mechanisms). You can find the complete version at [1.js](./1.js).
+For this example, we will use MulticastDNS for automatic Peer Discovery. This example is based the previous examples found in [Discovery Mechanisms](../discovery-mechanisms). You can find the complete version at [1.js](./1.js).
 
-Getting PubSub is super simple, all you have to do is require the FloodSub module and pass it in a libp2p node, once you have done that you can start subscribing and publishing in any topic.
+Using PubSub is super simple, all you have to do is start a libp2p node, PubSub will be enabled by default.
 
 ```JavaScript
-const FloodSub = require('libp2p-floodsub')
-
-const fs1 = new FloodSub(node1)
-const fs2 = new FloodSub(node2)
-
 series([
-  (cb) => fs1.start(cb),
-  (cb) => fs2.start(cb),
   (cb) => node1.once('peer:discovery', (peer) => node1.dial(peer, cb)),
   (cb) => setTimeout(cb, 500)
 ], (err) => {
   if (err) { throw err }
 
-  fs2.on('news', (msg) => console.log(msg.from, msg.data.toString()))
-  fs2.subscribe('news')
-
-  setInterval(() => {
-    fs1.publish('news', Buffer.from('Bird bird bird, bird is the word!'))
-  }, 1000)
+  // Subscribe to the topic 'news'
+  node1.pubsub.subscribe('news',
+    (msg) => console.log(msg.from, msg.data.toString()),
+    () => {
+      setInterval(() => {
+        // Publish the message on topic 'news'
+        node2.pubsub.publish(
+          'news',
+          Buffer.from('Bird bird bird, bird is the word!'),
+          () => {}
+        )
+      }, 1000)
+    }
+  )
 })
 ```
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p",
-  "version": "0.16.2",
+  "version": "0.19.0",
   "description": "JavaScript base class for libp2p bundles",
   "main": "src/index.js",
   "scripts": {
@@ -38,40 +38,41 @@
   "homepage": "https://github.com/libp2p/js-libp2p",
   "dependencies": {
     "async": "^2.6.0",
-    "libp2p-ping": "~0.6.0",
-    "libp2p-switch": "~0.36.0",
-    "mafmt": "^3.0.2",
+    "libp2p-floodsub": "^0.14.1",
+    "libp2p-ping": "~0.6.1",
+    "libp2p-switch": "~0.37.3",
+    "mafmt": "^4.0.0",
     "multiaddr": "^3.0.2",
     "peer-book": "~0.5.4",
-    "peer-id": "~0.10.5",
+    "peer-id": "~0.10.6",
     "peer-info": "~0.11.6"
   },
   "devDependencies": {
-    "aegir": "^12.4.0",
+    "aegir": "^13.0.6",
     "chai": "^4.1.2",
-    "cids": "~0.5.2",
+    "cids": "~0.5.3",
     "dirty-chai": "^2.0.1",
     "electron-webrtc": "~0.3.0",
-    "libp2p-circuit": "~0.1.4",
-    "libp2p-kad-dht": "~0.6.3",
+    "libp2p-circuit": "~0.1.5",
+    "libp2p-kad-dht": "~0.9.0",
     "libp2p-mdns": "~0.9.2",
-    "libp2p-multiplex": "~0.5.1",
+    "libp2p-mplex": "~0.6.0",
     "libp2p-railing": "~0.7.1",
-    "libp2p-secio": "~0.9.1",
+    "libp2p-secio": "~0.9.3",
     "libp2p-spdy": "~0.11.0",
-    "libp2p-tcp": "~0.11.5",
-    "libp2p-webrtc-star": "~0.13.3",
-    "libp2p-websockets": "~0.10.4",
-    "libp2p-websocket-star": "~0.7.2",
-    "libp2p-websocket-star-rendezvous": "~0.2.2",
+    "libp2p-tcp": "~0.11.6",
+    "libp2p-webrtc-star": "~0.13.4",
+    "libp2p-websockets": "~0.10.5",
+    "libp2p-websocket-star": "~0.7.7",
+    "libp2p-websocket-star-rendezvous": "~0.2.3",
     "lodash.times": "^4.3.2",
     "pre-commit": "^1.2.2",
     "pull-goodbye": "0.0.2",
     "pull-serializer": "~0.3.2",
-    "pull-stream": "^3.6.1",
+    "pull-stream": "^3.6.2",
     "safe-buffer": "^5.1.1",
-    "sinon": "^4.2.2",
-    "wrtc": "0.0.65"
+    "sinon": "^4.4.6",
+    "wrtc": "0.0.67"
   },
   "contributors": [
     "Chris Bratlien <chrisbratlien@gmail.com>",
@@ -94,6 +95,7 @@
     "Richard Littauer <richard.littauer@gmail.com>",
     "Ryan Bell <ryan@piing.net>",
     "Tiago Alves <alvesjtiago@gmail.com>",
+    "Zane Starr <zcstarr@gmail.com>",
     "greenkeeperio-bot <support@greenkeeper.io>",
     "mayerwin <mayerwin@users.noreply.github.com>",
     "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com>"

src/error-messages.js

@@ -0,0 +1,3 @@
+'use strict'
+
+exports.NOT_STARTED_YET = 'The libp2p node is not started yet'

src/index.js

@@ -14,6 +14,7 @@ const Ping = require('libp2p-ping')
 const peerRouting = require('./peer-routing')
 const contentRouting = require('./content-routing')
 const dht = require('./dht')
+const pubsub = require('./pubsub')
 const getPeerInfo = require('./get-peer-info')
 
 exports = module.exports
@@ -33,7 +34,8 @@ class Node extends EventEmitter {
 
     this._isStarted = false
 
-    this.switch = new Switch(this.peerInfo, this.peerBook)
+    this.switch = new Switch(this.peerInfo, this.peerBook, _options.switch)
+    this.stats = this.switch.stats
 
     // Attach stream multiplexers
     if (this.modules.connection && this.modules.connection.muxer) {
@@ -89,6 +91,7 @@ class Node extends EventEmitter {
     this.peerRouting = peerRouting(this)
     this.contentRouting = contentRouting(this)
     this.dht = dht(this)
+    this.pubsub = pubsub(this)
 
     this._getPeerInfo = getPeerInfo(this)
 
@@ -149,17 +152,29 @@ class Node extends EventEmitter {
         cb()
       },
       (cb) => {
-        // TODO: chicken-and-egg problem:
+        // TODO: chicken-and-egg problem #1:
         // have to set started here because DHT requires libp2p is already started
         this._isStarted = true
         if (this._dht) {
-          return this._dht.start(cb)
+          this._dht.start(cb)
+        } else {
+          cb()
         }
-        cb()
       },
+      (cb) => {
+        // TODO: chicken-and-egg problem #2:
+        // have to set started here because FloodSub requires libp2p is already started
+        if (this._options !== false) {
+          this._floodSub.start(cb)
+        } else {
+          cb()
+        }
+      },
+
       (cb) => {
         // detect which multiaddrs we don't have a transport for and remove them
         const multiaddrs = this.peerInfo.multiaddrs.toArray()
+
         transports.forEach((transport) => {
           multiaddrs.forEach((multiaddr) => {
             if (!multiaddr.toString().match(/\/p2p-circuit($|\/)/) &&
@@ -188,6 +203,11 @@ class Node extends EventEmitter {
     }
 
     series([
+      (cb) => {
+        if (this._floodSub.started) {
+          this._floodSub.stop(cb)
+        }
+      },
       (cb) => {
         if (this._dht) {
           return this._dht.stop(cb)

src/pubsub.js

@@ -0,0 +1,93 @@
+'use strict'
+
+const setImmediate = require('async/setImmediate')
+const NOT_STARTED_YET = require('./error-messages').NOT_STARTED_YET
+const FloodSub = require('libp2p-floodsub')
+
+module.exports = (node) => {
+  const floodSub = new FloodSub(node)
+
+  node._floodSub = floodSub
+
+  return {
+    subscribe: (topic, options, handler, callback) => {
+      if (typeof options === 'function') {
+        callback = handler
+        handler = options
+        options = {}
+      }
+
+      if (!node.isStarted() && !floodSub.started) {
+        return setImmediate(() => callback(new Error(NOT_STARTED_YET)))
+      }
+
+      function subscribe (cb) {
+        if (floodSub.listenerCount(topic) === 0) {
+          floodSub.subscribe(topic)
+        }
+
+        floodSub.on(topic, handler)
+        setImmediate(cb)
+      }
+
+      subscribe(callback)
+    },
+
+    unsubscribe: (topic, handler) => {
+      if (!node.isStarted() && !floodSub.started) {
+        throw new Error(NOT_STARTED_YET)
+      }
+
+      floodSub.removeListener(topic, handler)
+
+      if (floodSub.listenerCount(topic) === 0) {
+        floodSub.unsubscribe(topic)
+      }
+    },
+
+    publish: (topic, data, callback) => {
+      if (!node.isStarted() && !floodSub.started) {
+        return setImmediate(() => callback(new Error(NOT_STARTED_YET)))
+      }
+
+      if (!Buffer.isBuffer(data)) {
+        return setImmediate(() => callback(new Error('data must be a Buffer')))
+      }
+
+      floodSub.publish(topic, data)
+
+      setImmediate(() => callback())
+    },
+
+    ls: (callback) => {
+      if (!node.isStarted() && !floodSub.started) {
+        return setImmediate(() => callback(new Error(NOT_STARTED_YET)))
+      }
+
+      const subscriptions = Array.from(floodSub.subscriptions)
+
+      setImmediate(() => callback(null, subscriptions))
+    },
+
+    peers: (topic, callback) => {
+      if (!node.isStarted() && !floodSub.started) {
+        return setImmediate(() => callback(new Error(NOT_STARTED_YET)))
+      }
+
+      if (typeof topic === 'function') {
+        callback = topic
+        topic = null
+      }
+
+      const peers = Array.from(floodSub.peers.values())
+        .filter((peer) => topic ? peer.topics.has(topic) : true)
+        .map((peer) => peer.info.id.toB58String())
+
+      setImmediate(() => callback(null, peers))
+    },
+
+    setMaxListeners (n) {
+      return floodSub.setMaxListeners(n)
+    }
+  }
+}

test/node.js

@@ -4,7 +4,9 @@ require('./base')
 require('./transports.node')
 require('./stream-muxing.node')
 require('./peer-discovery.node')
+require('./pubsub.node')
 require('./peer-routing.node')
 require('./content-routing.node')
 require('./circuit-relay.node')
 require('./multiaddr-trim')
+require('./stats')

test/pubsub.node.js

@@ -0,0 +1,87 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 8] */
+
+'use strict'
+
+const chai = require('chai')
+chai.use(require('dirty-chai'))
+const expect = chai.expect
+const parallel = require('async/parallel')
+const waterfall = require('async/waterfall')
+const _times = require('lodash.times')
+const utils = require('./utils/node')
+const createNode = utils.createNode
+
+function startTwo (callback) {
+  const tasks = _times(2, () => (cb) => {
+    createNode('/ip4/0.0.0.0/tcp/0', {
+      mdns: false
+    }, (err, node) => {
+      expect(err).to.not.exist()
+      node.start((err) => cb(err, node))
+    })
+  })
+
+  parallel(tasks, (err, nodes) => {
+    expect(err).to.not.exist()
+
+    nodes[0].dial(nodes[1].peerInfo, (err) => callback(err, nodes))
+  })
+}
+
+function stopTwo (nodes, callback) {
+  parallel([
+    (cb) => nodes[0].stop(cb),
+    (cb) => nodes[1].stop(cb)
+  ], callback)
+}
+
+// There is a vast test suite on PubSub through js-ipfs
+// https://github.com/ipfs/interface-ipfs-core/blob/master/js/src/pubsub.js
+// and libp2p-floodsub itself
+// https://github.com/libp2p/js-libp2p-floodsub/tree/master/test
+// TODO: consider if all or some of those should come here
+describe('.pubsub', () => {
+  describe('.pubsub on (default)', (done) => {
+    it('start two nodes and send one message', (done) => {
+      waterfall([
+        (cb) => startTwo(cb),
+        (nodes, cb) => {
+          const data = Buffer.from('test')
+          nodes[0].pubsub.subscribe('pubsub',
+            (msg) => {
+              expect(msg.data).to.eql(data)
+              cb(null, nodes)
+            },
+            (err) => {
+              expect(err).to.not.exist()
+              setTimeout(() => nodes[1].pubsub.publish('pubsub', data, (err) => {
+                expect(err).to.not.exist()
+              }), 500)
+            }
+          )
+        },
+        (nodes, cb) => stopTwo(nodes, cb)
+      ], done)
+    })
+  })
+
+  describe('.pubsub off', () => {
+    it('fail to use pubsub if disabled', (done) => {
+      createNode('/ip4/0.0.0.0/tcp/0', {
+        mdns: false,
+        pubsub: false
+      }, (err, node) => {
+        expect(err).to.not.exist()
+
+        node.pubsub.subscribe('news',
+          (msg) => {},
+          (err) => {
+            expect(err).to.exist()
+            done()
+          }
+        )
+      })
+    })
+  })
+})

test/stats.js

@@ -0,0 +1,24 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+chai.use(require('dirty-chai'))
+const expect = chai.expect
+
+const createNode = require('./utils/node').createNode
+
+describe('libp2p', (done) => {
+  it('has stats', () => {
+    createNode('/ip4/127.0.0.1/tcp/0', {
+      mdns: false,
+      dht: true
+    }, (err, node) => {
+      expect(err).to.not.exist()
+      node.start((err) => {
+        expect(err).to.not.exist()
+        expect(node.stats).to.exist()
+        node.stop(done)
+      })
+    })
+  })
+})

test/stream-muxing.node.js

@@ -60,14 +60,14 @@ describe('stream muxing', () => {
     ], done)
   })
 
-  it('multiplex only', (done) => {
+  it('mplex only', (done) => {
     let nodeA
     let nodeB
 
     function setup (callback) {
       parallel([
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['multiplex']
+          muxer: ['mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeA = node
@@ -75,7 +75,7 @@ describe('stream muxing', () => {
           node.start(cb)
         }),
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['multiplex']
+          muxer: ['mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeB = node
@@ -92,7 +92,7 @@ describe('stream muxing', () => {
     ], done)
   })
 
-  it('spdy + multiplex', function (done) {
+  it('spdy + mplex', function (done) {
     this.timeout(5000)
 
     let nodeA
@@ -101,7 +101,7 @@ describe('stream muxing', () => {
     function setup (callback) {
       parallel([
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['spdy', 'multiplex']
+          muxer: ['spdy', 'mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeA = node
@@ -109,7 +109,7 @@ describe('stream muxing', () => {
           node.start(cb)
         }),
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['spdy', 'multiplex']
+          muxer: ['spdy', 'mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeB = node
@@ -126,7 +126,7 @@ describe('stream muxing', () => {
     ], done)
   })
 
-  it('spdy + multiplex switched order', function (done) {
+  it('spdy + mplex switched order', function (done) {
     this.timeout(5 * 1000)
 
     let nodeA
@@ -135,7 +135,7 @@ describe('stream muxing', () => {
     function setup (callback) {
       parallel([
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['spdy', 'multiplex']
+          muxer: ['spdy', 'mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeA = node
@@ -143,7 +143,7 @@ describe('stream muxing', () => {
           node.start(cb)
         }),
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['multiplex', 'spdy']
+          muxer: ['mplex', 'spdy']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeB = node
@@ -177,7 +177,7 @@ describe('stream muxing', () => {
           node.start(cb)
         }),
         (cb) => createNode('/ip4/0.0.0.0/tcp/0', {
-          muxer: ['multiplex']
+          muxer: ['mplex']
         }, (err, node) => {
           expect(err).to.not.exist()
           nodeB = node

test/transports.browser.js

@@ -48,12 +48,12 @@ describe('transports', () => {
       })
     })
 
-    it('create libp2pNode with multiplex only', (done) => {
+    it('create libp2pNode with mplex only', (done) => {
       PeerInfo.create((err, peerInfo) => {
         expect(err).to.not.exist()
 
-        const b = new Node(peerInfo, null, { muxer: ['multiplex'] })
-        expect(b.modules.connection.muxer).to.eql([require('libp2p-multiplex')])
+        const b = new Node(peerInfo, null, { muxer: ['mplex'] })
+        expect(b.modules.connection.muxer).to.eql([require('libp2p-mplex')])
         done()
       })
     })

test/utils/bundle.browser.js

@@ -4,7 +4,7 @@ const WS = require('libp2p-websockets')
 const WebRTCStar = require('libp2p-webrtc-star')
 const WebSocketStar = require('libp2p-websocket-star')
 const spdy = require('libp2p-spdy')
-const multiplex = require('libp2p-multiplex')
+const mplex = require('libp2p-mplex')
 const secio = require('libp2p-secio')
 const Railing = require('libp2p-railing')
 const libp2p = require('../..')
@@ -17,8 +17,8 @@ function mapMuxers (list) {
     switch (pref.trim().toLowerCase()) {
       case 'spdy':
         return spdy
-      case 'multiplex':
-        return multiplex
+      case 'mplex':
+        return mplex
       default:
         throw new Error(pref + ' muxer not available')
     }
@@ -29,7 +29,7 @@ function getMuxers (options) {
   if (options) {
     return mapMuxers(options)
   } else {
-    return [multiplex, spdy]
+    return [mplex, spdy]
   }
 }
 

test/utils/bundle.node.js

@@ -6,7 +6,7 @@ const WS = require('libp2p-websockets')
 const Railing = require('libp2p-railing')
 const spdy = require('libp2p-spdy')
 const KadDHT = require('libp2p-kad-dht')
-const multiplex = require('libp2p-multiplex')
+const mplex = require('libp2p-mplex')
 const secio = require('libp2p-secio')
 const libp2p = require('../..')
 
@@ -17,7 +17,7 @@ function mapMuxers (list) {
     }
     switch (pref.trim().toLowerCase()) {
       case 'spdy': return spdy
-      case 'multiplex': return multiplex
+      case 'mplex': return mplex
       default:
         throw new Error(pref + ' muxer not available')
     }
@@ -31,7 +31,7 @@ function getMuxers (muxers) {
   } else if (muxers) {
     return mapMuxers(muxers)
   } else {
-    return [multiplex, spdy]
+    return [mplex, spdy]
   }
 }
 

test/utils/node.js

@@ -22,7 +22,7 @@ function createNode (multiaddrs, options, callback) {
   }
 
   waterfall([
-    (cb) => PeerId.create({ bits: 1024 }, cb),
+    (cb) => PeerId.create({ bits: 512 }, cb),
     (peerId, cb) => PeerInfo.create(peerId, cb),
     (peerInfo, cb) => {
       multiaddrs.map((ma) => peerInfo.multiaddrs.add(ma))