fix: better error handling for REST transport (#1431)

* fix: better error handling for REST transport

* fix: change import of mkdirp

* fix: PR feedback

package.json

@@ -33,7 +33,6 @@
   },
   "devDependencies": {
     "@compodoc/compodoc": "1.1.19",
-    "@types/mkdirp": "^1.0.2",
     "@types/mocha": "^9.0.0",
     "@types/ncp": "^2.0.1",
     "@types/node": "^18.0.0",

src/fallbackServiceStub.ts

@@ -34,9 +34,9 @@ export interface FallbackServiceStub {
   // Compatible with gRPC service stub
   [method: string]: (
     request: {},
-    options: {},
-    metadata: {},
-    callback: (err?: Error, response?: {} | undefined) => void
+    options?: {},
+    metadata?: {},
+    callback?: (err?: Error, response?: {} | undefined) => void
   ) => StreamArrayParser | {cancel: () => void};
 }
 
@@ -83,10 +83,12 @@ export function generateServiceStub(
   for (const [rpcName, rpc] of Object.entries(rpcs)) {
     serviceStub[rpcName] = (
       request: {},
-      options: {[name: string]: string},
-      _metadata: {},
-      callback: Function
+      options?: {[name: string]: string},
+      _metadata?: {} | Function,
+      callback?: Function
     ) => {
+      options ??= {};
+
       // We cannot use async-await in this function because we need to return the canceller object as soon as possible.
       // Using plain old promises instead.
 
@@ -103,7 +105,9 @@ export function generateServiceStub(
       } catch (err) {
         // we could not encode parameters; pass error to the callback
         // and return a no-op canceler object.
-        callback(err);
+        if (callback) {
+          callback(err);
+        }
         return {
           cancel() {},
         };
@@ -171,7 +175,7 @@ export function generateServiceStub(
             ])
               .then(([ok, buffer]: [boolean, Buffer | ArrayBuffer]) => {
                 const response = responseDecoder(rpc, ok, buffer);
-                callback(null, response);
+                callback!(null, response);
               })
               .catch((err: Error) => {
                 if (!cancelRequested || err.name !== 'AbortError') {
@@ -180,14 +184,27 @@ export function generateServiceStub(
                       callback(err);
                     }
                     streamArrayParser.emit('error', err);
-                  } else {
+                  } else if (callback) {
                     callback(err);
+                  } else {
+                    throw err;
                   }
                 }
               });
           }
         })
-        .catch((err: unknown) => callback(err));
+        .catch((err: unknown) => {
+          if (rpc.responseStream) {
+            if (callback) {
+              callback(err);
+            }
+            streamArrayParser.emit('error', err);
+          } else if (callback) {
+            callback(err);
+          } else {
+            throw err;
+          }
+        });
 
       if (rpc.responseStream) {
         return streamArrayParser;

test/unit/regapic.ts

@@ -14,20 +14,19 @@
  * limitations under the License.
  */
 
-/* xslint-disable @typescript-eslint/ban-ts-comment */
-/* xslint-disable no-undef */
-
 import * as assert from 'assert';
 import {describe, it, afterEach, before} from 'mocha';
 import * as nodeFetch from 'node-fetch';
 import * as protobuf from 'protobufjs';
 import * as path from 'path';
 import * as sinon from 'sinon';
+import * as stream from 'stream';
 import echoProtoJson = require('../fixtures/echo.json');
 import {GrpcClient} from '../../src/fallback';
 import * as transcoding from '../../src/transcoding';
 import {OAuth2Client} from 'google-auth-library';
 import {GrpcClientOptions} from '../../src';
+import {StreamArrayParser} from '../../src/streamArrayParser';
 
 const authClient = {
   async getRequestHeaders() {
@@ -112,6 +111,75 @@ describe('REGAPIC', () => {
     });
   });
 
+  it('should make a streaming request', done => {
+    const requestObject = {content: 'test content'};
+    const responseObject = [{content: 'test'}, {content: 'content'}];
+    const responseObjectJson = JSON.stringify(responseObject, null, '  ');
+    const responseStream = new stream.Readable();
+    responseStream.push(responseObjectJson.slice(0, 10));
+    responseStream.push(responseObjectJson.slice(10));
+    responseStream.push(null);
+    // incomplete types for nodeFetch, so...
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    sinon.stub(nodeFetch, 'Promise' as any).returns(
+      Promise.resolve({
+        ok: true,
+        body: responseStream,
+      })
+    );
+
+    gaxGrpc.createStub(echoService, stubOptions).then(echoStub => {
+      const stream = echoStub.expand(
+        requestObject,
+        {},
+        {},
+        () => {}
+      ) as StreamArrayParser;
+      const results: {}[] = [];
+      stream.on('data', data => {
+        results.push(data);
+      });
+      stream.on('error', done);
+      stream.on('end', () => {
+        assert.deepStrictEqual(results, responseObject);
+        done();
+      });
+    });
+  });
+
+  it('should handle fetch failure', done => {
+    const requestObject = {content: 'test-content'};
+    sinon
+      // incomplete types for nodeFetch, so...
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      .stub(nodeFetch, 'Promise' as any)
+      .returns(Promise.reject(new Error('Fetch error')));
+
+    gaxGrpc.createStub(echoService, stubOptions).then(echoStub => {
+      echoStub.echo(requestObject, {}, {}, (err?: {}) => {
+        assert.strictEqual((err as Error).message, 'Fetch error');
+        done();
+      });
+    });
+  });
+
+  it('should handle streaming request failure', done => {
+    const requestObject = {content: 'test content'};
+    sinon
+      // incomplete types for nodeFetch, so...
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      .stub(nodeFetch, 'Promise' as any)
+      .returns(Promise.reject(new Error('Fetch error')));
+
+    gaxGrpc.createStub(echoService, stubOptions).then(echoStub => {
+      const stream = echoStub.expand(requestObject) as StreamArrayParser;
+      stream.on('error', err => {
+        assert.strictEqual((err as Error).message, 'Fetch error');
+        done();
+      });
+    });
+  });
+
   describe('should support enum conversion in proto message', () => {
     it('should support enum conversion in proto message response', done => {
       const requestObject = {name: 'shelves/shelf-name'};

test/unit/streamArrayParser.ts

@@ -132,7 +132,7 @@ describe('Parse REST stream array', () => {
     });
   });
 
-  it('should assign defaul value if the service response is not valid protobuf specific JSON', done => {
+  it('should assign default value if the service response is not valid protobuf specific JSON', done => {
     const expectedResults = [
       {
         name: 'Not Valid Name Message',

tools/prepublish.ts

@@ -19,7 +19,7 @@ import * as path from 'path';
 // Note: the following three imports will be all gone when we support Node.js 16+.
 // But until then, we'll use these modules.
 import * as rimraf from 'rimraf';
-import * as mkdirp from 'mkdirp';
+import mkdirp from 'mkdirp';
 import * as ncp from 'ncp';
 import {promisify} from 'util';