feat: emit 'response' with every page (#887)

`Bigtable.listAppProfiles` supports pagination and may return a unique value [`failedLocations`](https://github.com/googleapis/googleapis/blob/94fe3637559a257634d7b47a15bb8d976daff788/google/bigtable/admin/v2/bigtable_instance_admin.proto#L541).
Currently [streamified](https://github.com/googleapis/gax-nodejs/blob/eb7aa446dabf420d65831cd151b8b3076e087c71/src/paginationCalls/pageDescriptor.ts#L63) version of the paginated call does not expose `failedLocations` to user.

This PR offers to emit a decorated error explaining that resources from some locations are currently not available.
The error is emitted after all available resources have been `push`-ed to stream.

src/paginationCalls/pageDescriptor.ts

@@ -60,11 +60,18 @@ export class PageDescriptor implements Descriptor {
     const maxResults = 'maxResults' in options ? options.maxResults : -1;
     let pushCount = 0;
     let started = false;
-    function callback(err: Error | null, resources: Array<{}>, next: {}) {
+    function callback(
+      err: Error | null,
+      resources: Array<{}>,
+      next: {},
+      apiResp: {}
+    ) {
       if (err) {
         stream.emit('error', err);
         return;
       }
+      // emit full api response with every page.
+      stream.emit('response', apiResp);
       for (let i = 0; i < resources.length; ++i) {
         if (ended(stream)) {
           return;

test/unit/pagedIteration.ts

@@ -270,6 +270,19 @@ describe('paged iteration', () => {
       );
     });
 
+    it('emits response event with apiCall', done => {
+      const onResponse = sinon.spy();
+      // @ts-ignore incomplete option
+      const stream = descriptor.createStream(apiCall, {}, null);
+      stream.on('response', onResponse);
+      streamChecker(
+        stream,
+        () => assert.strictEqual(onResponse.callCount, pagesToStream + 1),
+        done,
+        0
+      );
+    });
+
     it('stops in the middle', done => {
       // @ts-ignore incomplete options
       const stream = descriptor.createStream(apiCall, {}, null);
@@ -306,15 +319,17 @@ describe('paged iteration', () => {
 
     it('caps the elements by maxResults', done => {
       const onData = sinon.spy();
+      const onResponse = sinon.spy();
       const stream =
         // @ts-ignore incomplete options
         descriptor.createStream(apiCall, {}, {maxResults: pageSize * 2 + 2});
-      stream.on('data', onData);
+      stream.on('data', onData).on('response', onResponse);
       streamChecker(
         stream,
         () => {
           assert.strictEqual(spy.callCount, 3);
           assert.strictEqual(onData.callCount, pageSize * 2 + 2);
+          assert.strictEqual(onResponse.callCount, spy.callCount);
         },
         done,
         0
@@ -375,5 +390,49 @@ describe('paged iteration', () => {
         })
         .on('error', done);
     });
+
+    describe('response', () => {
+      let nums: number[] = [];
+      let nextPageToken: number | undefined = undefined;
+      function sendResponse(
+        request: {pageToken?: number},
+        metadata: {},
+        options: {},
+        callback: APICallback
+      ) {
+        const pageToken = request.pageToken || 0;
+        if (pageToken >= pageSize * pagesToStream) {
+          nums = [];
+          nextPageToken = undefined;
+        } else {
+          nums = new Array(pageSize);
+          for (let i = 0; i < pageSize; i++) {
+            nums[i] = pageToken + i;
+          }
+          nextPageToken = pageToken + pageSize;
+        }
+        callback(null, {nums, nextPageToken});
+      }
+
+      it('should emit response object', done => {
+        const spy = sinon.spy(sendResponse);
+        const apiCall = util.createApiCall(spy, createOptions);
+        function onResponse(response: {}) {
+          assert.deepStrictEqual(response, {nums, nextPageToken});
+        }
+        const onResponseSpy = sinon.spy(onResponse);
+        // @ts-ignore incomplete options
+        const stream = descriptor.createStream(apiCall, {}, null);
+        stream.on('response', onResponseSpy);
+        streamChecker(
+          stream,
+          () => {
+            assert.strictEqual(onResponseSpy.callCount, spy.callCount);
+          },
+          done,
+          0
+        );
+      });
+    });
   });
 });