Don't emit end while waiting for pipes to clear

Fixes #506

lib/base.js

@@ -182,17 +182,26 @@ class Base extends MiniPass {
       this.options.failures = failures
 
     this.onbeforeend()
-    if (!this.flowing && this.buffer.length) {
+    // if we're piping, and buffered, then it means we need to hold off
+    // on emitting 'end' and calling ondone() until the pipes clear out.
+    if (this.pipes.length && this.buffer.length)
       super.end()
-    } else {
+    else
       this.emit('end')
-    }
-    this.ondone()
   }
 
   onbeforeend () {}
   ondone () {}
 
+  emit (ev, data) {
+    if (ev === 'end') {
+      const ret = super.emit(ev, data)
+      this.ondone()
+      return ret
+    } else
+      return super.emit(ev, data)
+  }
+
   setupParser (options) {
     this.parser = new Parser({
       bail: this.bail,

test/base.js

@@ -247,3 +247,29 @@ t.test('oncomplete', t => {
     b.parser.end(c[0])
   }), c[2])))
 })
+
+t.test('pipes backing up', t => {
+  const MiniPass = require('minipass')
+  const mp = new MiniPass({ encoding: 'utf8' })
+  const b = new Base({})
+  b.pipe(mp)
+  let flushed = false
+  let ended = false
+  b.on('end', () => {
+    t.notOk(flushed, 'not ending before flushing the stream')
+    t.notOk(ended, 'not ended more than once')
+    ended = true
+  })
+  const tapdata = 'TAP version 13\n1..1\nok\n'
+  b.parser.end(tapdata)
+  setTimeout(() => {
+    let data = ''
+    let c = ''
+    while (c = mp.read()) data += c
+    flushed = true
+    t.equal(data, tapdata)
+    // pipes should have flushed now
+    t.ok(ended, 'ended')
+    t.end()
+  })
+})

test/regression-pipe-backup.js

@@ -0,0 +1,10 @@
+'use strict'
+
+// https://github.com/tapjs/node-tap/pull/506
+
+const t = require('../')
+
+t.plan(5000)
+for (let i = 1; i <= 5000; i++) {
+  t.pass(i)
+}