fix: ETOOMANYSTREAMS 😢

Author: JGAntunes

src/core/components/ping.js

@@ -4,13 +4,15 @@ const promisify = require('promisify-es6')
 const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
 const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
-const Readable = require('readable-stream').Readable
+const pull = require('pull-stream/pull')
+const take = require('pull-stream/throughs/take')
+const Pushable = require('pull-pushable')
+const ndjson = require('pull-ndjson')
 
 function getPacket (msg) {
   // Default msg
   const basePacket = {Success: false, Time: 0, Text: ''}
-  // ndjson
-  return `${JSON.stringify(Object.assign({}, basePacket, msg))}\n`
+  return Object.assign({}, basePacket, msg)
 }
 
 module.exports = function ping (self) {
@@ -19,10 +21,15 @@ module.exports = function ping (self) {
       return cb(new Error(OFFLINE_ERROR))
     }
 
-    const source = new Readable({
-      read: function () {}
+    const source = Pushable(function (err) {
+      console.log('stream closed!', err)
     })
 
+    const response = pull(
+      source,
+      ndjson.serialize()
+    )
+
     let peer
     try {
       peer = self._libp2pNode.peerBook.get(peerId)
@@ -34,21 +41,33 @@ module.exports = function ping (self) {
     }
 
     self._libp2pNode.ping(peer, (err, p) => {
+      if (err) {
+        console.log('ERROR', err)
+        return source.abort(err)
+      }
       let packetCount = 0 
       let totalTime = 0
       source.push(getPacket({Success: true, Text: `PING ${peerId}`}))
       p.on('ping', (time) => {
+        console.log('ON PING')
         source.push(getPacket({ Success: true, Time: time }))
         totalTime += time
         packetCount++
+        console.log(packetCount, count)
         if (packetCount >= count) {
           const average = totalTime/count
           p.stop()
           source.push(getPacket({ Success: false, Text: `Average latency: ${average}ms`}))
-          source.push(null)
+          source.end()
         }
       })
+      console.log('Setup handler')
+      p.on('error', (err) => {
+        console.log('ERROR BATATA', err)
+        source.abort(err)
+      })
     })
-    cb(null, source)
+    
+    cb(null, response)
   })
 }

src/http/api/resources/ping.js

@@ -3,6 +3,8 @@
 const Joi = require('joi')
 const boom = require('boom')
 const toStream = require('pull-stream-to-stream')
+const PassThrough = require('readable-stream').PassThrough
+const pump = require('pump')
 
 exports = module.exports
 
@@ -24,13 +26,18 @@ exports.get = {
     // Default count to 10
     const count = request.query.n || request.query.count || 10
 
-    ipfs.ping(peerId, count, (err, sourceStream) => {
+    ipfs.ping(peerId, count, (err, pullStream) => {
       if (err) {
         return reply(boom.badRequest(err))
       }
-      console.log(sourceStream)
-
-      return reply(sourceStream).type('application/json').header('x-chunked-output', '1')
+      // Streams from pull-stream-to-stream don't seem to be compatible
+      // with the stream2 readable interface
+      // see: https://github.com/hapijs/hapi/blob/c23070a3de1b328876d5e64e679a147fafb04b38/lib/response.js#L533
+      // and: https://github.com/pull-stream/pull-stream-to-stream/blob/e436acee18b71af8e71d1b5d32eee642351517c7/index.js#L28
+      const responseStream = toStream.source(pullStream)
+      const stream2 = new PassThrough()
+      pump(responseStream, stream2)
+      return reply(stream2).type('application/json').header('x-chunked-output', '1')
     })
   }
 }