Merge pull request #58 from alanshaw/patch-1

fix: parameter to _onPeerDisconnected

Author: wemeetagain

package.json

@@ -58,6 +58,7 @@
     "lodash": "^4.17.15",
     "mocha": "^6.2.1",
     "p-times": "^2.1.0",
+    "p-wait-for": "^3.1.0",
     "promisify-es6": "^1.0.3",
     "sinon": "^7.5.0"
   },

src/pubsub.js

@@ -98,7 +98,7 @@ class BasicPubSub extends Pubsub {
    * @override
    * @param {string} idB58Str peer id string in base58
    * @param {Connection} conn connection
-   * @param {PeerInfo} peer peer info
+   * @param {Peer} peer PubSub peer
    * @returns {void}
    *
    */
@@ -117,7 +117,7 @@ class BasicPubSub extends Pubsub {
         }
       )
     } catch (err) {
-      this._onPeerDisconnected(peer, err)
+      this._onPeerDisconnected(peer.info, err)
     }
   }
 

test/pubsub.spec.js

@@ -6,10 +6,12 @@ const chai = require('chai')
 chai.use(require('dirty-chai'))
 const expect = chai.expect
 const sinon = require('sinon')
+const pWaitFor = require('p-wait-for')
 
 const { utils } = require('libp2p-pubsub')
 const {
   createGossipsub,
+  createPeerInfo,
   mockRegistrar
 } = require('./utils')
 
@@ -120,4 +122,34 @@ describe('Pubsub', () => {
       }, 500))
     })
   })
+
+  describe('process', () => {
+    it('should disconnect peer on stream error', async () => {
+      sinon.spy(gossipsub, '_onPeerDisconnected')
+
+      const peerInfo = await createPeerInfo()
+      const mockConn = {
+        newStream () {
+          return {
+            stream: {
+              sink: async source => {
+                for await (const _ of source) { // eslint-disable-line no-unused-vars
+                  // mock stream just swallows any data sent
+                }
+              },
+              source: (async function * () { // eslint-disable-line require-yield
+                // throw in a bit
+                await new Promise(resolve => setTimeout(resolve, 100))
+                throw new Error('boom')
+              })()
+            }
+          }
+        }
+      }
+
+      gossipsub._onPeerConnected(peerInfo, mockConn)
+
+      await pWaitFor(() => gossipsub._onPeerDisconnected.calledWith(peerInfo), { timeout: 1000 })
+    })
+  })
 })