feat: support async iterator for paging method

src/paginationCalls/pageDescriptor.ts

@@ -18,12 +18,22 @@ import * as ended from 'is-stream-ended';
 import {PassThrough, Transform} from 'stream';
 
 import {APICaller} from '../apiCaller';
-import {GaxCall, APICallback} from '../apitypes';
+import {
+  GaxCall,
+  APICallback,
+  SimpleCallbackFunction,
+  RequestType,
+  GaxCallPromise,
+} from '../apitypes';
 import {Descriptor} from '../descriptor';
 import {CallSettings} from '../gax';
 import {NormalApiCaller} from '../normalCalls/normalApiCaller';
 
 import {PagedApiCaller} from './pagedApiCaller';
+import * as call_1 from '../call';
+
+let resolveFunction: Function;
+let resolveRequest: Function;
 
 /**
  * A descriptor for methods that support pagination.
@@ -103,6 +113,102 @@ export class PageDescriptor implements Descriptor {
     return stream;
   }
 
+  // create async iterator(settings) => iterable
+  async(apiCall: GaxCall, request: {}, options: CallSettings): {} {
+    const iterable = this.createIterator(options);
+    const funcPromise =
+      typeof apiCall === 'function' ? Promise.resolve(apiCall) : apiCall;
+    funcPromise
+      .then((func: GaxCall) => {
+        this.resolveParams(request, func, options);
+      })
+      .catch(error => {
+        throw new Error(error);
+      });
+    return iterable;
+  }
+
+  createIterator(options: CallSettings): {} {
+    const responsePageTokenFieldName = this.responsePageTokenField;
+    const requestPageTokenFieldName = this.requestPageTokenField;
+    const asyncIterable = {
+      [Symbol.asyncIterator]() {
+        const funcPromise = new Promise((resolve, reject) => {
+          resolveFunction = resolve;
+        });
+        const requestPromise = new Promise((resolve, reject) => {
+          resolveRequest = resolve;
+        });
+        const cache: Array<{}> = [];
+        let nextPageRequest: RequestType = {};
+        let firstCall = true;
+        return {
+          async next() {
+            const ongoingCall = new call_1.OngoingCallPromise(options.promise);
+
+            const func = (await funcPromise) as SimpleCallbackFunction;
+            const request = (await requestPromise) as RequestType;
+            if (firstCall) {
+              ongoingCall.call(func, request);
+              const [
+                response,
+                nextRequest,
+                rawresponse,
+              ] = await ongoingCall.promise;
+              //@ts-ignore
+              cache.push(...response.responses.map(r => r.content));
+              //@ts-ignore
+              const pageToken = response[responsePageTokenFieldName];
+              if (pageToken) {
+                nextPageRequest = Object.assign({}, request);
+                nextPageRequest[requestPageTokenFieldName] = pageToken;
+              }
+              firstCall = false;
+              return Promise.resolve({done: false, value: cache.shift()});
+            } else {
+              if (cache.length > 0) {
+                const value = cache.shift();
+                return Promise.resolve({done: false, value});
+              } else if (nextPageRequest) {
+                ongoingCall.call(func, nextPageRequest);
+                const [
+                  response,
+                  nextRequest,
+                  rawResponse,
+                ] = await ongoingCall.promise;
+                //@ts-ignore
+                const pageToken = response[responsePageTokenFieldName];
+                if (pageToken) {
+                  nextPageRequest[requestPageTokenFieldName] = pageToken;
+                }
+                //@ts-ignore
+                else nextPageRequest = null;
+                //@ts-ignore
+                cache.push(...response.responses.map(r => r.content));
+                const value = cache.shift();
+                return Promise.resolve({done: false, value});
+              } else {
+                return Promise.resolve({done: true, value: -1});
+              }
+            }
+          },
+        };
+      },
+    };
+    return asyncIterable; // return iterable
+  }
+
+  resolveParams(request: RequestType, func: GaxCall, settings: CallSettings) {
+    if (settings.pageToken) {
+      request[this.requestPageTokenField] = settings.pageToken;
+    }
+    if (settings.pageSize) {
+      request[this.requestPageSizeField!] = settings.pageSize;
+    }
+    resolveRequest(request);
+    resolveFunction(func);
+  }
+
   getApiCaller(settings: CallSettings): APICaller {
     if (!settings.autoPaginate) {
       return new NormalApiCaller();

test/fixtures/google-gax-packaging-test-app/src/index.js

@@ -75,6 +75,7 @@ async function testShowcase() {
   await testEcho(grpcClient);
   await testExpand(grpcClient);
   await testPagedExpand(grpcClient);
+  await testPagedExpandAsync(grpcClient);
   await testCollect(grpcClient);
   await testChat(grpcClient);
   await testWait(grpcClient);
@@ -136,6 +137,20 @@ async function testPagedExpand(client) {
   assert.deepStrictEqual(words, result);
 }
 
+async function testPagedExpandAsync(client) {
+  const words = ['nobody', 'ever', 'reads', 'test', 'input'];
+  const request = {
+    content: words.join(' '),
+    pageSize: 2,
+  };
+  const response = [];
+  const iterable = client.pagedExpandAsync(request);
+  for await (const resource of iterable){
+    response.push(resource);
+  }
+  assert.deepStrictEqual(words, response);
+}
+
 async function testCollect(client) {
   const words = ['nobody', 'ever', 'reads', 'test', 'input'];
   const result = await new Promise((resolve, reject) => {

test/fixtures/google-gax-packaging-test-app/src/v1beta1/echo_client.js

@@ -197,8 +197,9 @@ class EchoClient {
       'wait',
       'pagedExpand',
     ];
+    this._innerCallPromise = {};
     for (const methodName of echoStubMethods) {
-      const innerCallPromise = echoStub.then(
+      this._innerCallPromise[methodName] = echoStub.then(
         stub => (...args) => {
           return stub[methodName].apply(stub, args);
         },
@@ -207,7 +208,7 @@ class EchoClient {
         }
       );
       this._innerApiCalls[methodName] = gaxModule.createApiCall(
-        innerCallPromise,
+        this._innerCallPromise[methodName],
         defaults[methodName],
         this._descriptors.page[methodName] ||
           this._descriptors.stream[methodName] ||
@@ -584,6 +585,13 @@ class EchoClient {
 
     return this._innerApiCalls.pagedExpand(request, options, callback);
   }
+
+  pagedExpandAsync(request, options) {
+      options = options || {};
+      request = request || {};
+      const callSettings = new gax.CallSettings(options);
+      return this._descriptors.page.pagedExpand.async(this._innerCallPromise['pagedExpand'], request, callSettings);
+  }
 }
 
 module.exports = EchoClient;

test/fixtures/google-gax-packaging-test-app/test/gapic-v1beta1.js

@@ -177,6 +177,45 @@ describe('EchoClient', () => {
       });
     });
 
+    it('invokes pagedExpand using async iterator', async () => {
+      const client = new showcaseModule.v1beta1.EchoClient({
+        credentials: {client_email: 'bogus', private_key: 'bogus'},
+        projectId: 'bogus',
+      });
+
+      // Mock request
+      const request = {};
+
+      // Mock response
+      const nextPageToken = '';
+      const responsesElement = {};
+      const expectedResponse = {
+        nextPageToken: nextPageToken,
+        responses: responsesElement,
+      };
+
+      client._descriptors.page.pagedExpand.async = (apiCall, request, options) => {
+        const asyncIterable = {
+          [Symbol.asyncIterator]() {
+            return {
+              async next(){
+                return Promise.resolve({done: true, value: -1});
+              }
+            }
+          }
+        }
+        return asyncIterable;
+      }
+
+      // test paging method by async iterator
+      const response = {};
+      const iterable = client.pagedExpandAsync(request);
+      for await (const resource of iterable){
+        response.push(resource);
+      }
+      assert.deepStrictEqual(response, expectedResponse.responses);
+    });
+
     it('invokes pagedExpand with error', done => {
       const client = new showcaseModule.v1beta1.EchoClient({
         credentials: {client_email: 'bogus', private_key: 'bogus'},

test/unit/pagedIteration.ts

@@ -24,6 +24,7 @@ import {APICallback, GaxCallPromise} from '../../src/apitypes';
 
 import * as util from './utils';
 import {Stream} from 'stream';
+import * as gax from '../../src/gax';
 
 describe('paged iteration', () => {
   const pageSize = 3;
@@ -194,6 +195,34 @@ describe('paged iteration', () => {
     });
   });
 
+  describe('use async iterator', () => {
+    // tslint:disable-next-line no-any
+    let spy: any;
+    let apiCall: GaxCallPromise;
+    beforeEach(() => {
+      spy = sinon.spy(func);
+      apiCall = util.createApiCall(spy, createOptions);
+    });
+
+    async function iterableChecker(
+      // tslint:disable-next-line no-any
+      iterable: any
+    ) {
+      let counter = 0;
+      for await (const resource of iterable) {
+        counter++;
+        if (counter === 10) break;
+      }
+      expect(counter).to.equal(10);
+    }
+    it('returns an iterable, count to 10', () => {
+      const settings = new gax.CallSettings(
+        (createOptions && createOptions.settings) || {}
+      );
+      iterableChecker(descriptor.async(apiCall, {}, settings));
+    });
+  });
+
   describe('stream conversion', () => {
     // tslint:disable-next-line no-any
     let spy: any;