fix: use unidirectional streams (#105)

* fix: use unidirectional streams * chore: use libp2p-pubsub@0.5.2
libp2p · Jun 4, 2020 · 96d3265 · 96d3265
1 parent 70dd732
commit 96d3265
Show file tree

Hide file tree

Showing 3 changed files with 122 additions and 2 deletions.
diff --git a/package.json b/package.json
@@ -60,7 +60,7 @@
     "debug": "^4.1.1",
     "it-length-prefixed": "^3.0.0",
     "it-pipe": "^1.0.1",
-    "libp2p-pubsub": "^0.5.0",
+    "libp2p-pubsub": "~0.5.2",
     "p-map": "^4.0.0",
     "peer-id": "~0.13.3",
     "protons": "^1.0.1",

diff --git a/test/2-nodes.spec.js b/test/2-nodes.spec.js
@@ -58,11 +58,14 @@ describe('basics between 2 nodes', () => {
     // Connect floodsub nodes
     before(async () => {
       const onConnectA = registrarRecordA[multicodec].onConnect
+      const onConnectB = registrarRecordB[multicodec].onConnect
+      const handleA = registrarRecordA[multicodec].handler
       const handleB = registrarRecordB[multicodec].handler
 
       // Notice peers of connection
       const [c0, c1] = ConnectionPair()
       await onConnectA(peerIdB, c0)
+      await onConnectB(peerIdA, c1)
 
       await handleB({
         protocol: multicodec,
@@ -72,6 +75,14 @@ describe('basics between 2 nodes', () => {
         }
       })
 
+      await handleA({
+        protocol: multicodec,
+        stream: c0.stream,
+        connection: {
+          remotePeer: peerIdB
+        }
+      })
+
       expect(fsA.peers.size).to.be.eql(1)
       expect(fsB.peers.size).to.be.eql(1)
     })
@@ -275,11 +286,28 @@ describe('basics between 2 nodes', () => {
       const dial = async () => {
         const onConnectA = registrarRecordA[multicodec].onConnect
         const onConnectB = registrarRecordB[multicodec].onConnect
+        const handleA = registrarRecordA[multicodec].handler
+        const handleB = registrarRecordB[multicodec].handler
 
         // Notice peers of connection
         const [c0, c1] = ConnectionPair()
         await onConnectA(peerIdB, c0)
+        await handleB({
+          protocol: multicodec,
+          stream: c1.stream,
+          connection: {
+            remotePeer: peerIdA
+          }
+        })
+
         await onConnectB(peerIdA, c1)
+        await handleA({
+          protocol: multicodec,
+          stream: c0.stream,
+          connection: {
+            remotePeer: peerIdB
+          }
+        })
       }
 
       await Promise.all([

diff --git a/test/multiple-nodes.spec.js b/test/multiple-nodes.spec.js
@@ -57,15 +57,46 @@ describe('multiple nodes (more than 2)', () => {
         const onConnectA = registrarRecordA[multicodec].onConnect
         const onConnectB = registrarRecordB[multicodec].onConnect
         const onConnectC = registrarRecordC[multicodec].onConnect
+        const handleA = registrarRecordA[multicodec].handler
+        const handleB = registrarRecordB[multicodec].handler
+        const handleC = registrarRecordC[multicodec].handler
 
         // Notice peers of connection
         const [d0, d1] = ConnectionPair()
         await onConnectA(peerIdB, d0)
+        await handleB({
+          protocol: multicodec,
+          stream: d1.stream,
+          connection: {
+            remotePeer: peerIdA
+          }
+        })
         await onConnectB(peerIdA, d1)
+        await handleA({
+          protocol: multicodec,
+          stream: d0.stream,
+          connection: {
+            remotePeer: peerIdB
+          }
+        })
 
         const [d2, d3] = ConnectionPair()
         await onConnectB(peerIdC, d2)
+        await handleC({
+          protocol: multicodec,
+          stream: d3.stream,
+          connection: {
+            remotePeer: peerIdB
+          }
+        })
         await onConnectC(peerIdB, d3)
+        await handleB({
+          protocol: multicodec,
+          stream: d2.stream,
+          connection: {
+            remotePeer: peerIdC
+          }
+        })
       })
 
       after(() => Promise.all([
@@ -267,23 +298,84 @@ describe('multiple nodes (more than 2)', () => {
         const onConnectC = registrarRecordC[multicodec].onConnect
         const onConnectD = registrarRecordD[multicodec].onConnect
         const onConnectE = registrarRecordE[multicodec].onConnect
+        const handleA = registrarRecordA[multicodec].handler
+        const handleB = registrarRecordB[multicodec].handler
+        const handleC = registrarRecordC[multicodec].handler
+        const handleD = registrarRecordD[multicodec].handler
+        const handleE = registrarRecordE[multicodec].handler
 
         // Notice peers of connection
         const [d0, d1] = ConnectionPair() // A <-> B
         await onConnectA(peerIdB, d0)
+        await handleB({
+          protocol: multicodec,
+          stream: d1.stream,
+          connection: {
+            remotePeer: peerIdA
+          }
+        })
         await onConnectB(peerIdA, d1)
+        await handleA({
+          protocol: multicodec,
+          stream: d0.stream,
+          connection: {
+            remotePeer: peerIdB
+          }
+        })
 
         const [d2, d3] = ConnectionPair() // B <-> C
         await onConnectB(peerIdC, d2)
+        await handleC({
+          protocol: multicodec,
+          stream: d3.stream,
+          connection: {
+            remotePeer: peerIdB
+          }
+        })
         await onConnectC(peerIdB, d3)
+        await handleB({
+          protocol: multicodec,
+          stream: d2.stream,
+          connection: {
+            remotePeer: peerIdC
+          }
+        })
 
         const [d4, d5] = ConnectionPair() // C <-> D
         await onConnectC(peerIdD, d4)
+        await handleD({
+          protocol: multicodec,
+          stream: d5.stream,
+          connection: {
+            remotePeer: peerIdC
+          }
+        })
         await onConnectD(peerIdC, d5)
+        await handleC({
+          protocol: multicodec,
+          stream: d4.stream,
+          connection: {
+            remotePeer: peerIdD
+          }
+        })
 
-        const [d6, d7] = ConnectionPair() // C <-> D
+        const [d6, d7] = ConnectionPair() // D <-> E
         await onConnectD(peerIdE, d6)
+        await handleE({
+          protocol: multicodec,
+          stream: d7.stream,
+          connection: {
+            remotePeer: peerIdD
+          }
+        })
         await onConnectE(peerIdD, d7)
+        await handleD({
+          protocol: multicodec,
+          stream: d6.stream,
+          connection: {
+            remotePeer: peerIdE
+          }
+        })
       })
 
       after(() => Promise.all([