fix: Stop Duplicate Response Handlers on Retries (#502)

* fix: Stop duplicate reponse handlers on retries

* style: lint

src/index.ts

@@ -761,12 +761,6 @@ export class Upload extends Pumpify {
       },
     });
 
-    // This should be 'once' as `startUploading` can be called again for
-    // multi chunk uploads and each request would have its own response.
-    this.once('response', resp => {
-      responseReceived = true;
-      this.responseHandler(resp);
-    });
     let headers: GaxiosOptions['headers'] = {};
 
     // If using multiple chunk upload, set appropriate header
@@ -791,7 +785,11 @@ export class Upload extends Pumpify {
     };
 
     try {
-      await this.makeRequestStream(reqOpts);
+      const resp = await this.makeRequestStream(reqOpts);
+      if (resp) {
+        responseReceived = true;
+        this.responseHandler(resp);
+      }
     } catch (err) {
       const e = err as Error;
       this.destroy(e);
@@ -983,7 +981,7 @@ export class Upload extends Pumpify {
     return res;
   }
 
-  private async makeRequestStream(reqOpts: GaxiosOptions): GaxiosPromise {
+  private async makeRequestStream(reqOpts: GaxiosOptions) {
     const controller = new AbortController();
     const errorCallback = () => controller.abort();
     this.once('error', errorCallback);
@@ -1002,10 +1000,10 @@ export class Upload extends Pumpify {
       reqOpts
     );
     const res = await this.authClient.request(combinedReqOpts);
-    this.onResponse(res);
+    const successfulRequest = this.onResponse(res);
     this.removeListener('error', errorCallback);
 
-    return res;
+    return successfulRequest ? res : null;
   }
 
   private restart() {

test/test.ts

@@ -13,7 +13,7 @@ import * as mockery from 'mockery';
 import * as nock from 'nock';
 import * as path from 'path';
 import * as sinon from 'sinon';
-import {PassThrough, Readable} from 'stream';
+import {Readable} from 'stream';
 
 import {ApiError, CreateUriCallback, PROTOCOL_REGEX} from '../src';
 import {GaxiosOptions, GaxiosError, GaxiosResponse} from 'gaxios';
@@ -896,7 +896,7 @@ describe('gcs-resumable-upload', () => {
 
   describe('#startUploading', () => {
     beforeEach(() => {
-      up.makeRequestStream = async () => new PassThrough();
+      up.makeRequestStream = async () => null;
       up.upstreamChunkBuffer = Buffer.alloc(16);
     });
 
@@ -968,14 +968,6 @@ describe('gcs-resumable-upload', () => {
       up.startUploading();
     });
 
-    it("should setup a 'response' listener", async () => {
-      assert.equal(up.eventNames().includes('response'), false);
-
-      await up.startUploading();
-
-      assert.equal(up.eventNames().includes('response'), true);
-    });
-
     it('should destroy the stream if the request failed', done => {
       const error = new Error('Error.');
       up.on('error', (e: Error) => {
@@ -1687,14 +1679,27 @@ describe('gcs-resumable-upload', () => {
       up.makeRequestStream(REQ_OPTS);
     });
 
-    it('should return the response', async () => {
-      const response = {};
+    it('should return the response if successful', async () => {
+      const response = {some: 'response'};
       up.authClient = {
         request: async () => response,
       };
+      up.onResponse = () => true;
+
       const stream = await up.makeRequestStream(REQ_OPTS);
       assert.strictEqual(stream, response);
     });
+
+    it('should return `null` if the response is unsuccessful', async () => {
+      const response = {some: 'response'};
+      up.authClient = {
+        request: async () => response,
+      };
+      up.onResponse = () => false;
+
+      const stream = await up.makeRequestStream(REQ_OPTS);
+      assert.strictEqual(stream, null);
+    });
   });
 
   describe('#restart', () => {
@@ -2251,7 +2256,7 @@ describe('gcs-resumable-upload', () => {
           let dataReceived = 0;
           let chunkWritesInRequest = 0;
 
-          await new Promise(resolve => {
+          const res = await new Promise(resolve => {
             opts.body.on('data', (data: Buffer) => {
               dataReceived += data.byteLength;
               overallDataReceived += data.byteLength;
@@ -2261,14 +2266,14 @@ describe('gcs-resumable-upload', () => {
             opts.body.on('end', () => {
               requests.push({dataReceived, opts, chunkWritesInRequest});
 
-              up.emit('response', {
+              resolve({
                 status: 200,
                 data: {},
               });
-
-              resolve(null);
             });
           });
+
+          return res;
         };
 
         up.on('error', done);
@@ -2390,7 +2395,7 @@ describe('gcs-resumable-upload', () => {
           let dataReceived = 0;
           let chunkWritesInRequest = 0;
 
-          await new Promise(resolve => {
+          const res = await new Promise(resolve => {
             opts.body.on('data', (data: Buffer) => {
               dataReceived += data.byteLength;
               overallDataReceived += data.byteLength;
@@ -2405,23 +2410,23 @@ describe('gcs-resumable-upload', () => {
                   ? overallDataReceived - 1
                   : 0;
 
-                up.emit('response', {
+                resolve({
                   status: RESUMABLE_INCOMPLETE_STATUS_CODE,
                   headers: {
                     range: `bytes=0-${lastByteReceived}`,
                   },
                   data: {},
                 });
               } else {
-                up.emit('response', {
+                resolve({
                   status: 200,
                   data: {},
                 });
               }
-
-              resolve(null);
             });
           });
+
+          return res;
         };
 
         up.on('error', done);