chore: release version v0.12.7

fix: call cb in close after all transport are closed

README.md

@@ -109,6 +109,12 @@ handle a new protocol.
 - `protocol`
 - `handler` - function called when we receive a dial on `protocol. Signature must be `function (conn) {}`
 
+### `swarm.unhandle(protocol)`
+
+unhandle a protocol.
+
+- `protocol`
+
 ### `swarm.close(callback)`
 
 close all the listeners and muxers.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p-swarm",
-  "version": "0.10.6",
+  "version": "0.12.7",
   "description": "libp2p swarm implementation in JavaScript",
   "main": "lib/index.js",
   "jsnext:main": "src/index.js",
@@ -11,8 +11,8 @@
     "test:node": "gulp test:node",
     "test:browser": "gulp test:browser",
     "release": "gulp release",
-    "release-minor": "gulp release --minor",
-    "release-major": "gulp release --major",
+    "release-minor": "gulp release --type minor",
+    "release-major": "gulp release --type major",
     "coverage": "gulp coverage",
     "coverage-publish": "aegir-coverage publish"
   },
@@ -40,13 +40,13 @@
     "bl": "^1.1.2",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
-    "aegir": "^2.1.1",
+    "aegir": "^3.0.0",
     "gulp": "^3.9.1",
     "istanbul": "^0.4.3",
     "libp2p-multiplex": "^0.2.1",
     "libp2p-spdy": "^0.3.1",
-    "libp2p-tcp": "^0.4.0",
-    "libp2p-websockets": "^0.3.2",
+    "libp2p-tcp": "^0.5.0",
+    "libp2p-websockets": "^0.4.1",
     "multiaddr": "^1.4.0",
     "peer-id": "^0.6.6",
     "peer-info": "^0.6.2",
@@ -54,6 +54,7 @@
     "stream-pair": "^1.0.3"
   },
   "dependencies": {
+    "async": "^2.0.0-rc.4",
     "babel-runtime": "^6.6.1",
     "duplex-passthrough": "github:diasdavid/duplex-passthrough",
     "ip-address": "^5.8.0",
@@ -74,8 +75,8 @@
     "David Dias <daviddias.p@gmail.com>",
     "David Dias <mail@daviddias.me>",
     "Francisco Baio Dias <xicombd@gmail.com>",
+    "Friedel Ziegelmayer <dignifiedquire@gmail.com>",
     "Pau Ramon Revilla <masylum@gmail.com>",
-    "Richard Littauer <richard.littauer@gmail.com>",
-    "dignifiedquire <dignifiedquire@gmail.com>"
+    "Richard Littauer <richard.littauer@gmail.com>"
   ]
 }

src/identify.js

@@ -41,7 +41,7 @@ exports.exec = (rawConn, muxer, peerInfo, callback) => {
 
       pbs.on('identify', (msg) => {
         if (msg.observedAddr.length > 0) {
-          peerInfo.multiaddr.addSafe(msg.observedAddr)
+          peerInfo.multiaddr.addSafe(multiaddr(msg.observedAddr))
         }
 
         const peerId = Id.createFromPubKey(msg.publicKey)
@@ -59,7 +59,7 @@ exports.exec = (rawConn, muxer, peerInfo, callback) => {
         protocolVersion: 'na',
         agentVersion: 'na',
         publicKey: peerInfo.id.pubKey,
-        listenAddrs: peerInfo.multiaddrs.map((mh) => { return mh.buffer }),
+        listenAddrs: peerInfo.multiaddrs.map((mh) => mh.buffer),
         observedAddr: obsMultiaddr ? obsMultiaddr.buffer : new Buffer('')
       })
 
@@ -70,15 +70,15 @@ exports.exec = (rawConn, muxer, peerInfo, callback) => {
 }
 
 exports.handler = (peerInfo, swarm) => {
-  return function (conn) {
+  return (conn) => {
     // 1. receive incoming observed info about me
     // 2. update my own information (on peerInfo)
     // 3. send back what I see from the other (get from swarm.muxedConns[incPeerID].conn.getObservedAddrs()
     var pbs = pbStream()
 
-    pbs.on('identify', function (msg) {
+    pbs.on('identify', (msg) => {
       if (msg.observedAddr.length > 0) {
-        peerInfo.multiaddr.addSafe(msg.observedAddr)
+        peerInfo.multiaddr.addSafe(multiaddr(msg.observedAddr))
       }
 
       const peerId = Id.createFromPubKey(msg.publicKey)
@@ -89,9 +89,7 @@ exports.handler = (peerInfo, swarm) => {
         protocolVersion: 'na',
         agentVersion: 'na',
         publicKey: peerInfo.id.pubKey,
-        listenAddrs: peerInfo.multiaddrs.map(function (ma) {
-          return ma.buffer
-        }),
+        listenAddrs: peerInfo.multiaddrs.map((ma) => ma.buffer),
         observedAddr: obsMultiaddr ? obsMultiaddr.buffer : new Buffer('')
       })
       pbs.finalize()

src/index.js

@@ -1,12 +1,17 @@
 'use strict'
 
+const async = require('async')
 const multistream = require('multistream-select')
 const identify = require('./identify')
 const DuplexPassThrough = require('duplex-passthrough')
 const contains = require('lodash.contains')
+const util = require('util')
+const EE = require('events').EventEmitter
 
 exports = module.exports = Swarm
 
+util.inherits(Swarm, EE)
+
 function Swarm (peerInfo) {
   if (!(this instanceof Swarm)) {
     return new Swarm(peerInfo)
@@ -144,7 +149,23 @@ function Swarm (peerInfo) {
     // for listening
     this.handle(muxer.multicodec, (conn) => {
       const muxedConn = muxer(conn, true)
+
+      var peerIdForConn
+
       muxedConn.on('stream', (conn) => {
+        function gotId () {
+          if (peerIdForConn) {
+            conn.peerId = peerIdForConn
+            connHandler(conn)
+          } else {
+            setTimeout(gotId, 100)
+          }
+        }
+
+        if (this.identify) {
+          return gotId()
+        }
+
         connHandler(conn)
       })
 
@@ -154,9 +175,18 @@ function Swarm (peerInfo) {
           if (err) {
             return console.log('Identify exec failed', err)
           }
+
+          peerIdForConn = pi.id
           this.muxedConns[pi.id.toB58String()] = {}
           this.muxedConns[pi.id.toB58String()].muxer = muxedConn
           this.muxedConns[pi.id.toB58String()].conn = conn // to be able to extract addrs
+
+          self.emit('peer-mux-established', pi)
+
+          muxedConn.on('close', () => {
+            delete self.muxedConns[pi.id.toB58String()]
+            self.emit('peer-mux-closed', pi)
+          })
         })
       }
     })
@@ -183,15 +213,19 @@ function Swarm (peerInfo) {
 
   // higher level (public) API
   this.dial = (pi, protocol, callback) => {
-    var pt = null
     if (typeof protocol === 'function') {
       callback = protocol
       protocol = null
-    } else {
-      pt = new DuplexPassThrough()
     }
 
+    if (!callback) {
+      callback = function noop () {}
+    }
+
+    const pt = new DuplexPassThrough()
+
     const b58Id = pi.id.toB58String()
+
     if (!this.muxedConns[b58Id]) {
       if (!this.conns[b58Id]) {
         attemptDial(pi, (err, conn) => {
@@ -284,14 +318,25 @@ function Swarm (peerInfo) {
             self.muxedConns[b58Id].muxer = muxedConn
             self.muxedConns[b58Id].conn = conn
 
+            self.emit('peer-mux-established', pi)
+
+            muxedConn.on('close', () => {
+              delete self.muxedConns[pi.id.toB58String()]
+              self.emit('peer-mux-closed', pi)
+            })
+
             // in case identify is on
-            muxedConn.on('stream', connHandler)
+            muxedConn.on('stream', (conn) => {
+              conn.peerId = pi.id
+              connHandler(conn)
+            })
 
             cb(null, muxedConn)
           })
         })
       }
     }
+
     function openConnInMuxedConn (muxer, cb) {
       cb(muxer.newStream())
     }
@@ -305,6 +350,7 @@ function Swarm (peerInfo) {
           }
 
           pt.wrapStream(conn)
+          pt.peerId = pi.id
           callback(null, pt)
         })
       })
@@ -315,20 +361,25 @@ function Swarm (peerInfo) {
     this.protocols[protocol] = handler
   }
 
-  this.close = (callback) => {
-    var count = 0
+  this.unhandle = (protocol, handler) => {
+    if (this.protocols[protocol]) {
+      delete this.protocols[protocol]
+    }
+  }
 
+  this.close = (callback) => {
     Object.keys(this.muxedConns).forEach((key) => {
       this.muxedConns[key].muxer.end()
     })
 
-    Object.keys(this.transports).forEach((key) => {
-      this.transports[key].close(() => {
-        if (++count === Object.keys(this.transports).length) {
-          callback()
-        }
-      })
-    })
+    async.each(
+      Object.keys(this.transports),
+      (key, cb) => this.transports[key].close(cb),
+      () => {
+        // Ignoring close errors
+        callback()
+      }
+    )
   }
 }
 

test/01-transport-tcp.node.js

@@ -46,9 +46,17 @@ describe('transport - tcp', function () {
     function ready () {
       if (++count === 2) {
         expect(peerA.multiaddrs.length).to.equal(1)
-        expect(peerA.multiaddrs[0]).to.deep.equal(multiaddr('/ip4/127.0.0.1/tcp/9888'))
+        expect(
+          peerA.multiaddrs[0].equals(multiaddr('/ip4/127.0.0.1/tcp/9888'))
+        ).to.be.equal(
+          true
+        )
         expect(peerB.multiaddrs.length).to.equal(1)
-        expect(peerB.multiaddrs[0]).to.deep.equal(multiaddr('/ip4/127.0.0.1/tcp/9999'))
+        expect(
+          peerB.multiaddrs[0].equals(multiaddr('/ip4/127.0.0.1/tcp/9999'))
+        ).to.be.equal(
+          true
+        )
         done()
       }
     }
@@ -124,7 +132,11 @@ describe('transport - tcp', function () {
 
     function ready () {
       expect(peer.multiaddrs.length).to.equal(1)
-      expect(peer.multiaddrs[0]).to.deep.equal(multiaddr('/ip4/0.0.0.0/tcp/9050'))
+      expect(
+        peer.multiaddrs[0].equals(multiaddr('/ip4/0.0.0.0/tcp/9050'))
+      ).to.be.equal(
+        true
+      )
       swarm.close(done)
     }
   })

test/03-transport-websockets.node.js

@@ -46,9 +46,17 @@ describe('transport - websockets', function () {
     function ready () {
       if (++count === 2) {
         expect(peerA.multiaddrs.length).to.equal(1)
-        expect(peerA.multiaddrs[0]).to.deep.equal(multiaddr('/ip4/127.0.0.1/tcp/9888/websockets'))
+        expect(
+          peerA.multiaddrs[0].equals(multiaddr('/ip4/127.0.0.1/tcp/9888/websockets'))
+        ).to.be.equal(
+          true
+        )
         expect(peerB.multiaddrs.length).to.equal(1)
-        expect(peerB.multiaddrs[0]).to.deep.equal(multiaddr('/ip4/127.0.0.1/tcp/9999/websockets'))
+        expect(
+          peerB.multiaddrs[0].equals(multiaddr('/ip4/127.0.0.1/tcp/9999/websockets'))
+        ).to.equal(
+          true
+        )
         done()
       }
     }

test/05-muxing-spdy.node.js

@@ -59,10 +59,10 @@ describe('stream muxing with spdy (on TCP)', function () {
 
     swarmA.close(closed)
     swarmB.close(closed)
-    swarmC.close(closed)
+    // swarmC.close(closed)
 
     function closed () {
-      if (++counter === 3) {
+      if (++counter === 2) {
         done()
       }
     }
@@ -128,4 +128,11 @@ describe('stream muxing with spdy (on TCP)', function () {
       }, 500)
     })
   })
+
+  it('close one end, make sure the other does not blow', (done) => {
+    swarmC.close(() => {
+      // to make sure it has time to propagate
+      setTimeout(done, 1000)
+    })
+  })
 })

test/08-swarm-without-muxing.node.js

@@ -103,4 +103,11 @@ describe('high level API - 1st without stream multiplexing (on TCP)', function (
       conn.on('end', done)
     })
   })
+
+  it('unhandle', (done) => {
+    const proto = '/bananas/1.0.0'
+    swarmA.unhandle(proto)
+    expect(swarmA.protocols[proto]).to.not.exist
+    done()
+  })
 })

test/09-swarm-with-muxing.node.js

@@ -49,7 +49,7 @@ describe('high level API - with everything mixed all together!', function () {
 
     swarmA.close(closed)
     swarmB.close(closed)
-    swarmC.close(closed)
+    // swarmC.close(closed)
     swarmD.close(closed)
     swarmE.close(closed)
 
@@ -135,6 +135,14 @@ describe('high level API - with everything mixed all together!', function () {
       conn.pipe(conn)
     })
 
+    swarmB.once('peer-mux-established', (peerInfo) => {
+      expect(peerInfo.id.toB58String()).to.equal(peerA.id.toB58String())
+    })
+
+    swarmA.once('peer-mux-established', (peerInfo) => {
+      expect(peerInfo.id.toB58String()).to.equal(peerB.id.toB58String())
+    })
+
     swarmA.dial(peerB, '/anona/1.0.0', (err, conn) => {
       expect(err).to.not.exist
       expect(Object.keys(swarmA.muxedConns).length).to.equal(1)
@@ -145,6 +153,14 @@ describe('high level API - with everything mixed all together!', function () {
     })
   })
 
+  it('dial from ws to ws no proto', (done) => {
+    swarmD.dial(peerE, (err) => {
+      expect(err).to.not.exist
+      expect(Object.keys(swarmD.muxedConns).length).to.equal(1)
+      done()
+    })
+  })
+
   it('dial from ws to ws', (done) => {
     swarmE.handle('/abacaxi/1.0.0', (conn) => {
       conn.pipe(conn)
@@ -182,11 +198,13 @@ describe('high level API - with everything mixed all together!', function () {
 
   it('dial from tcp+ws to tcp+ws', (done) => {
     swarmC.handle('/mamao/1.0.0', (conn) => {
+      expect(conn.peerId).to.exist
       conn.pipe(conn)
     })
 
     swarmA.dial(peerC, '/mamao/1.0.0', (err, conn) => {
       expect(err).to.not.exist
+      expect(conn.peerId).to.exist
       expect(Object.keys(swarmA.muxedConns).length).to.equal(2)
       conn.end()
 
@@ -194,4 +212,11 @@ describe('high level API - with everything mixed all together!', function () {
       conn.on('end', done)
     })
   })
+
+  it('close a muxer emits event', (done) => {
+    swarmC.close(() => {})
+    swarmA.once('peer-mux-closed', (peerInfo) => {
+      done()
+    })
+  })
 })