Ensure Readable can start flowing after Writable finish #671

docs/changelog.md

@@ -6,6 +6,10 @@ Requires libvips v8.4.2.
 
 #### v0.17.2 - TBD
 
+* Ensure Readable side of Stream can start flowing after Writable side has finished.
+  [#671](https://github.com/lovell/sharp/issues/671)
+  [@danhaller](https://github.com/danhaller)
+
 * Expose WebP alpha quality, lossless and near-lossless output options.
   [#685](https://github.com/lovell/sharp/pull/685)
   [@rnanwani](https://github.com/rnanwani)

lib/input.js

@@ -64,6 +64,12 @@ const _write = function _write (chunk, encoding, callback) {
   if (Array.isArray(this.options.input.buffer)) {
     /* istanbul ignore else */
     if (is.buffer(chunk)) {
+      if (this.options.input.buffer.length === 0) {
+        const that = this;
+        this.on('finish', function () {
+          that.streamInFinished = true;
+        });
+      }
       this.options.input.buffer.push(chunk);
       callback();
     } else {

lib/output.js

@@ -378,9 +378,9 @@ const _pipeline = function _pipeline (callback) {
     // output=stream
     if (this._isStreamInput()) {
       // output=stream, input=stream
-      this.on('finish', function () {
-        that._flattenBufferIn();
-        sharp.pipeline(that.options, function (err, data, info) {
+      if (this.streamInFinished) {
+        this._flattenBufferIn();
+        sharp.pipeline(this.options, function (err, data, info) {
           if (err) {
             that.emit('error', err);
           } else {
@@ -389,7 +389,20 @@ const _pipeline = function _pipeline (callback) {
           }
           that.push(null);
         });
-      });
+      } else {
+        this.on('finish', function () {
+          that._flattenBufferIn();
+          sharp.pipeline(that.options, function (err, data, info) {
+            if (err) {
+              that.emit('error', err);
+            } else {
+              that.emit('info', info);
+              that.push(data);
+            }
+            that.push(null);
+          });
+        });
+      }
     } else {
       // output=stream, input=file/buffer
       sharp.pipeline(this.options, function (err, data, info) {

test/unit/io.js

@@ -157,6 +157,28 @@ describe('Input/output', function () {
     readableButNotAnImage.pipe(writable);
   });
 
+  it('Readable side of Stream can start flowing after Writable side has finished', function (done) {
+    const readable = fs.createReadStream(fixtures.inputJpg);
+    const writable = fs.createWriteStream(fixtures.outputJpg);
+    writable.on('finish', function () {
+      sharp(fixtures.outputJpg).toBuffer(function (err, data, info) {
+        if (err) throw err;
+        assert.strictEqual(true, data.length > 0);
+        assert.strictEqual(data.length, info.size);
+        assert.strictEqual('jpeg', info.format);
+        assert.strictEqual(320, info.width);
+        assert.strictEqual(240, info.height);
+        fs.unlinkSync(fixtures.outputJpg);
+        done();
+      });
+    });
+    const pipeline = sharp().resize(320, 240);
+    readable.pipe(pipeline);
+    pipeline.on('finish', function () {
+      pipeline.pipe(writable);
+    });
+  });
+
   it('Sequential read, force JPEG', function (done) {
     sharp(fixtures.inputJpg)
       .sequentialRead()