pack backpressure (#114)

* backpressure: RFC

* tweak test

* trim whitespace

Co-authored-by: Mathias Buus <mathiasbuus@gmail.com>

pack.js

@@ -128,9 +128,10 @@ Pack.prototype.entry = function (header, buffer, callback) {
   if (Buffer.isBuffer(buffer)) {
     header.size = buffer.length
     this._encode(header)
-    this.push(buffer)
+    var ok = this.push(buffer)
     overflow(self, header.size)
-    process.nextTick(callback)
+    if (ok) process.nextTick(callback)
+    else this._drain = callback
     return new Void()
   }
 

test/pack.js

@@ -3,6 +3,7 @@ var tar = require('../index')
 var fixtures = require('./fixtures')
 var concat = require('concat-stream')
 var fs = require('fs')
+var Writable = require('readable-stream').Writable
 
 test('one-file', function (t) {
   t.plan(2)
@@ -190,3 +191,45 @@ test('unicode', function (t) {
     t.deepEqual(data, fs.readFileSync(fixtures.UNICODE_TAR))
   }))
 })
+
+test('backpressure', function (t) {
+  var slowWritable = new Writable({ highWaterMark: 1 })
+  slowWritable._write = (chunk, enc, next) => {
+    setImmediate(next)
+  }
+
+  var pack = tar.pack()
+  var later = false
+
+  setImmediate(() => {
+    later = true
+  })
+
+  pack.pipe(slowWritable)
+
+  slowWritable.on('finish', () => t.end())
+  pack.on('end', () => t.ok(later))
+
+  var i = 0
+  var next = () => {
+    if (++i < 25) {
+      var header = {
+        name: `file${i}.txt`,
+        mtime: new Date(1387580181000),
+        mode: parseInt('644', 8),
+        uname: 'maf',
+        gname: 'staff',
+        uid: 501,
+        gid: 20
+      }
+
+      var buffer = Buffer.alloc(1024)
+
+      pack.entry(header, buffer, next)
+    } else {
+      pack.finalize()
+    }
+  }
+
+  next()
+})