fix: stop stream after first pong received (#545)

When connecting to go-IPFS from a webworker, the stream opened by
the ping protocol is never closed.

The change here uses `take` to only receive one buffer from the
remote node before closing the stream.

Author: achingbrain

src/ping/index.js

@@ -8,7 +8,7 @@ const errCode = require('err-code')
 const crypto = require('libp2p-crypto')
 const pipe = require('it-pipe')
 const { toBuffer } = require('it-buffer')
-const { collect } = require('streaming-iterables')
+const { collect, take } = require('streaming-iterables')
 
 const { PROTOCOL, PING_LENGTH } = require('./constants')
 
@@ -29,6 +29,7 @@ async function ping (node, peer) {
   const [result] = await pipe(
     [data],
     stream,
+    stream => take(1, stream),
     toBuffer,
     collect
   )

test/core/ping.node.js

@@ -6,9 +6,11 @@ chai.use(require('dirty-chai'))
 const { expect } = chai
 
 const pTimes = require('p-times')
+const pipe = require('it-pipe')
 
 const peerUtils = require('../utils/creators/peer')
 const baseOptions = require('../utils/base-options')
+const { PROTOCOL } = require('../../src/ping/constants')
 
 describe('ping', () => {
   let nodes
@@ -32,4 +34,40 @@ describe('ping', () => {
     const averageLatency = latencies.reduce((p, c) => p + c, 0) / latencies.length
     expect(averageLatency).to.be.a('Number')
   })
+
+  it('only waits for the first response to arrive', async () => {
+    nodes[1].handle(PROTOCOL, async ({ connection, stream }) => {
+      let firstInvocation = true
+
+      await pipe(
+        stream,
+        function (stream) {
+          const output = {
+            [Symbol.asyncIterator]: () => output,
+            next: async () => {
+              if (firstInvocation) {
+                firstInvocation = false
+
+                for await (const data of stream) {
+                  return {
+                    value: data,
+                    done: false
+                  }
+                }
+              } else {
+                return new Promise() // never resolve
+              }
+            }
+          }
+
+          return output
+        },
+        stream
+      )
+    })
+
+    const latency = await nodes[0].ping(nodes[1].peerInfo)
+
+    expect(latency).to.be.a('Number')
+  })
 })