fix: undefined stream.destroy call

To end streaming in case of error, we were calling an unofficial method of the stream API which was removed and does not exist in the version of node we use. The method is re-added officially in node v.8 but until we upgrade we need to destroy the streams manually, by pushing null for readables and calling stream.end() for writables.

lib/s3middleware/azureHelpers/SubStreamInterface.js

@@ -15,6 +15,7 @@ class SubStreamInterface {
         this._lengthCounter = 0;
         this._subPartIndex = 0;
         this._currentStream = new stream.PassThrough();
+        this._streamingAborted = false;
     }
 
     /**
@@ -51,12 +52,16 @@ class SubStreamInterface {
      * @return {undefined}
      */
     stopStreaming(piper) {
+        this._streamingAborted = true;
         if (piper) {
             piper.unpipe();
-            piper.destroy();
+            piper.push(null);
         }
-        this._sourceStream.destroy();
-        this._currentStream.destroy();
+        this._sourceStream.pause();
+        this._sourceStream.push(null);
+        this._sourceStream.end();
+        this._currentStream.push(null);
+        this._currentStream.end();
     }
 
     /**
@@ -111,6 +116,10 @@ class SubStreamInterface {
      * @return {undefined}
      */
     write(chunk) {
+        if (this._streamingAborted) {
+            // don't write
+            return;
+        }
         const ready = this._currentStream.write(chunk);
 
         if (!ready) {

lib/s3middleware/azureHelpers/mpuUtils.js

@@ -151,9 +151,9 @@ dataStoreName, log, cb) => {
         'putting multiple parts');
 
     resultsCollector.on('error', (err, subPartIndex) => {
-        streamInterface.stopStreaming(request);
         log.error(`Error putting subpart to Azure: ${subPartIndex}`,
             { error: err.message, dataStoreName });
+        streamInterface.stopStreaming(request);
         if (err.code === 'ContainerNotFound') {
             return cb(errors.NoSuchBucket);
         }

tests/unit/s3middleware/azureHelpers/SubStreamingInterface.js

@@ -0,0 +1,53 @@
+const assert = require('assert');
+const stream = require('stream');
+const SubStreamInterface =
+    require('../../../../lib/s3middleware/azureHelpers/SubStreamInterface');
+
+describe('s3middleware SubStreamInterface.stopStreaming()', () => {
+    const emittedFinished = {
+        sourceStream: false,
+        currentStream: false,
+    };
+    const data = Buffer.alloc(100);
+    let dataMarker = 0;
+    const mockRequest = new stream.Readable({
+        read: () => {
+            if (dataMarker >= data.length) {
+                return mockRequest.push(null);
+            }
+            mockRequest.push(data.slice(dataMarker, dataMarker + 1));
+            dataMarker += 1;
+            return undefined;
+        },
+    });
+    const sourceStream = new stream.PassThrough();
+    const subStreamInterface = new SubStreamInterface(sourceStream);
+    sourceStream.on('finish', () => {
+        emittedFinished.sourceStream = true;
+    });
+    subStreamInterface._currentStream.on('finish', () => {
+        emittedFinished.currentStream = true;
+    });
+    it('should stop streaming data and end all streams', done => {
+        sourceStream.on('data', chunk => {
+            const currentLength = subStreamInterface.getLengthCounter();
+            if (currentLength === 10) {
+                Object.keys(emittedFinished).forEach(key => {
+                    assert.strictEqual(emittedFinished[key], false);
+                });
+                assert.strictEqual(mockRequest._readableState.pipesCount, 1);
+                return subStreamInterface.stopStreaming(mockRequest);
+            }
+            return subStreamInterface.write(chunk);
+        });
+        mockRequest.pipe(sourceStream);
+        setTimeout(() => {
+            Object.keys(emittedFinished).forEach(key => {
+                assert.strictEqual(emittedFinished[key], true);
+            });
+            assert.strictEqual(subStreamInterface.getLengthCounter(), 10);
+            assert.strictEqual(mockRequest._readableState.pipesCount, 0);
+            return done();
+        }, 1000);
+    });
+});