fix: dont blow up the stack

libp2p · Feb 7, 2019 · c041265 · c041265
1 parent d43efac
commit c041265
Show file tree

Hide file tree

Showing 3 changed files with 50 additions and 51 deletions.
diff --git a/package.json b/package.json
@@ -42,6 +42,7 @@
     "dirty-chai": "^2.0.1",
     "interface-connection": "^0.3.2",
     "lodash.defaults": "^4.2.0",
+    "looper": "^4.0.0",
     "pull-offset-limit": "^1.1.1",
     "pull-pair": "^1.1.0",
     "pull-pushable": "^2.2.0",

diff --git a/src/channel.js b/src/channel.js
@@ -2,6 +2,7 @@
 
 const pushable = require('pull-pushable')
 const defaults = require('lodash.defaults')
+const looper = require('looper')
 
 const consts = require('./consts')
 const EE = require('events')
@@ -50,10 +51,10 @@ class Channel extends EE {
       })
     }
 
-    // this._log('new channel', this._name)
+    this._log('new channel', this._name)
 
     this._msgs = pushable((err) => {
-      // this._log('source closed', err)
+      this._log('source closed', err)
       if (err && typeof err !== 'boolean') {
         setImmediate(() => this.emit('error', err))
       }
@@ -63,33 +64,33 @@ class Channel extends EE {
     this._source = this._msgs
 
     this.sink = (read) => {
-      const next = (end, data) => {
-        // this._log('sink', data)
-
-        // stream already ended
-        if (this._endedLocal) { return }
-
-        this._endedLocal = end || false
-
-        // source ended, close the stream
-        if (end === true) {
-          return this.endChan()
-        }
-
-        // source errored, reset stream
-        if (end || this._reset) {
-          this.resetChan()
-          this.emit('error', end || this._reset)
-          this.reset()
-          return
-        }
-
-        // just send
-        this.sendMsg(data)
-        return read(null, next)
-      }
+      const next = looper(() => {
+        read(null, (end, data) => {
+          // stream already ended
+          if (this._endedLocal) { return }
+
+          this._endedLocal = end || false
+
+          // source ended, close the stream
+          if (end === true) {
+            return this.endChan()
+          }
+
+          // source errored, reset stream
+          if (end || this._reset) {
+            this.resetChan()
+            this.emit('error', end || this._reset)
+            this.reset()
+            return
+          }
+
+          // just send
+          this.sendMsg(data)
+          next()
+        })
+      })
 
-      read(null, next)
+      next()
     }
   }
 
@@ -118,13 +119,13 @@ class Channel extends EE {
   }
 
   push (data) {
-    // this._log('push', data)
+    this._log('push', data)
     this._msgs.push(data)
   }
 
   // close for reading
   close (err) {
-    // this._log('close', err)
+    this._log('close', err)
     if (!this._endedRemote) {
       this._endedRemote = err || true
       this._msgs.end(this._endedRemote)
@@ -134,14 +135,12 @@ class Channel extends EE {
   }
 
   reset (err) {
-    // this._log('reset', err)
+    this._log('reset', err)
     this._reset = err || 'channel reset!'
     this.close(this._reset)
   }
 
   openChan () {
-    // this._log('openChan')
-
     if (this.open) { return } // chan already open
 
     let name
@@ -154,7 +153,7 @@ class Channel extends EE {
   }
 
   sendMsg (data) {
-    // this._log('sendMsg', data)
+    this._log('sendMsg')
 
     if (!this.open) {
       this.openChan()
@@ -168,7 +167,7 @@ class Channel extends EE {
   }
 
   endChan () {
-    // this._log('endChan')
+    this._log('endChan')
 
     if (!this.open) {
       return
@@ -181,7 +180,7 @@ class Channel extends EE {
   }
 
   resetChan () {
-    // this._log('endChan')
+    this._log('resetChan')
 
     if (!this.open) {
       return

diff --git a/src/index.js b/src/index.js
@@ -3,6 +3,7 @@
 const pull = require('pull-stream')
 const pushable = require('pull-pushable')
 const through = require('pull-through')
+const looper = require('looper')
 
 const defautls = require('lodash.defaults')
 
@@ -57,7 +58,7 @@ class Mplex extends EE {
     }
 
     this._chandata = pushable((err) => {
-      // this._log('chandata ended')
+      this._log('mplex ended')
       this._endedRemote = true
       this.close(err)
     })
@@ -82,15 +83,16 @@ class Mplex extends EE {
       }),
       coder.decode(),
       (read) => {
-        function next (end, data) {
-          if (self._endedLocal) { return }
-          if (end === true) { return self.close() }
-          if (end) { return self.reset(end) }
-          self._handle(data)
-          return read(null, next)
-        }
-
-        return read(null, next)
+        const next = looper(() => {
+          read(null, function (end, data) {
+            if (self._endedLocal) { return }
+            if (end === true) { return self.close() }
+            if (end) { return self.reset(end) }
+            self._handle(data)
+            next()
+          })
+        })
+        next()
       })
   }
 
@@ -99,7 +101,7 @@ class Mplex extends EE {
   }
 
   close (err) {
-    // this._log('close', err)
+    this._log('close', err)
 
     if (this.destroyed) { return }
 
@@ -132,14 +134,12 @@ class Mplex extends EE {
   }
 
   push (data) {
-    // this._log('push', data)
     if (data.data &&
       Buffer.byteLength(data.data) > this._maxMsgSize) {
       this._chandata.end(new Error('message too large!'))
     }
 
     this._chandata.push(data)
-    log('buffer', this._chandata.buffer)
   }
 
   _nextChanId () {
@@ -173,7 +173,6 @@ class Mplex extends EE {
     }
 
     id = typeof id === 'number' ? id : this._nextChanId(initiator)
-    // if (list.has(id)) {
     if (list[id]) {
       this.emit('error', new Error(`channel with id ${id} already exist!`))
       return
@@ -203,7 +202,7 @@ class Mplex extends EE {
   }
 
   _handle (msg) {
-    // this._log('_handle', msg)
+    this._log('_handle', msg)
     const { id, type, data } = msg
     switch (type) {
       case consts.type.NEW: {