fix: make async iteration work for gRPC-fallback; refactor the code (#765)

alexander-fenster · web-flow · commit 944c06b4225c · 2020-03-27T22:19:15.000-07:00
diff --git a/samples/package.json b/samples/package.json
@@ -3,7 +3,7 @@
   "license": "Apache-2.0",
   "author": "Google LLC",
   "engines": {
-    "node": ">=8"
+    "node": ">=10"
   },
   "repository": "googleapis/gax-nodejs",
   "private": true,
diff --git a/samples/pagination.js b/samples/pagination.js
@@ -67,6 +67,16 @@ async function main() {
   // Call it!
   const [resources] = await wrappedFunction({request: 'empty'});
   console.log(resources);
+
+  // Alternatively, call it using async iterators!
+  const iterable = pageDescriptor.asyncIterate(wrappedFunction, {
+    request: 'empty',
+  });
+  const asyncResources = [];
+  for await (const resource of iterable) {
+    asyncResources.push(resource);
+  }
+  console.log(asyncResources);
 }
 
 main().catch(console.error);
diff --git a/src/apitypes.ts b/src/apitypes.ts
@@ -42,7 +42,7 @@ export type NextPageRequestType = {
 } | null;
 export type RawResponseType = Operation | {} | null;
 export type ResultTuple = [
-  ResponseType,
+  ResponseType | [ResponseType],
   NextPageRequestType | undefined,
   RawResponseType | undefined
 ];
diff --git a/src/paginationCalls/pageDescriptor.ts b/src/paginationCalls/pageDescriptor.ts
@@ -18,18 +18,12 @@ import * as ended from 'is-stream-ended';
 import {PassThrough, Transform} from 'stream';
 
 import {APICaller} from '../apiCaller';
-import {
-  GaxCall,
-  APICallback,
-  SimpleCallbackFunction,
-  RequestType,
-} from '../apitypes';
+import {GaxCall, APICallback, RequestType, ResultTuple} from '../apitypes';
 import {Descriptor} from '../descriptor';
 import {CallSettings} from '../gax';
 import {NormalApiCaller} from '../normalCalls/normalApiCaller';
 
 import {PagedApiCaller} from './pagedApiCaller';
-import * as call from '../call';
 
 export interface ResponseType {
   [index: string]: string;
@@ -38,12 +32,10 @@ export interface ResponseType {
  * A descriptor for methods that support pagination.
  */
 export class PageDescriptor implements Descriptor {
-  resolveParams: Function;
   requestPageTokenField: string;
   responsePageTokenField: string;
   requestPageSizeField?: string;
   resourceField: string;
-  cache: Array<{}>;
 
   constructor(
     requestPageTokenField: string,
@@ -53,8 +45,6 @@ export class PageDescriptor implements Descriptor {
     this.requestPageTokenField = requestPageTokenField;
     this.responsePageTokenField = responsePageTokenField;
     this.resourceField = resourceField;
-    this.resolveParams = () => {};
-    this.cache = [];
   }
 
   /**
@@ -122,93 +112,49 @@ export class PageDescriptor implements Descriptor {
   asyncIterate(
     apiCall: GaxCall,
     request: RequestType,
-    options: CallSettings
+    options?: CallSettings
   ): AsyncIterable<{} | undefined> {
-    const iterable = this.createIterator();
-    const funcPromise =
-      typeof apiCall === 'function' ? Promise.resolve(apiCall) : apiCall;
-    funcPromise
-      .then((func: GaxCall) => {
-        this.makeCall(request, func, options);
-      })
-      .catch(error => {
-        throw new Error(error);
-      });
+    options = Object.assign({}, options, {autoPaginate: false});
+    const iterable = this.createIterator(apiCall, request, options);
     return iterable;
   }
 
-  createIterator(): AsyncIterable<{} | undefined> {
-    // eslint-disable-next-line @typescript-eslint/no-this-alias
-    const self = this;
+  createIterator(
+    apiCall: GaxCall,
+    request: RequestType,
+    options: CallSettings
+  ): AsyncIterable<{} | undefined> {
     const asyncIterable = {
       [Symbol.asyncIterator]() {
-        const paramPromise: Promise<[
-          RequestType,
-          SimpleCallbackFunction
-        ]> = new Promise(resolve => {
-          self.resolveParams = resolve;
-        });
-        let nextPageRequest: RequestType | null = {};
-        let firstCall = true;
+        let nextPageRequest: RequestType | null | undefined = request;
+        const cache: {}[] = [];
         return {
           async next() {
-            const ongoingCall = new call.OngoingCallPromise();
-            const [request, func] = await paramPromise;
-            if (self.cache.length > 0) {
+            if (cache.length > 0) {
               return Promise.resolve({
                 done: false,
-                value: self.cache.shift(),
+                value: cache.shift(),
               });
             }
-            if (!firstCall && !nextPageRequest) {
-              return Promise.resolve({done: true, value: undefined});
+            if (nextPageRequest) {
+              let result: {} | [ResponseType] | null;
+              [result, nextPageRequest] = (await apiCall(
+                nextPageRequest!,
+                options
+              )) as ResultTuple;
+              cache.push(...(result as ResponseType[]));
             }
-            nextPageRequest = await self.getNextPageRequest(
-              func,
-              firstCall ? request : nextPageRequest!,
-              ongoingCall
-            );
-            firstCall = false;
-            if (self.cache.length === 0) {
-              nextPageRequest = null;
+            if (cache.length === 0) {
               return Promise.resolve({done: true, value: undefined});
             }
-            return Promise.resolve({done: false, value: self.cache.shift()});
+            return Promise.resolve({done: false, value: cache.shift()});
           },
         };
       },
     };
     return asyncIterable;
   }
 
-  async getNextPageRequest(
-    func: SimpleCallbackFunction,
-    request: RequestType,
-    ongoingCall: call.OngoingCallPromise
-  ): Promise<RequestType | null> {
-    ongoingCall.call(func, request);
-    let nextPageRequest = null;
-    const [response] = await ongoingCall.promise;
-    const pageToken = (response as ResponseType)[this.responsePageTokenField];
-    if (pageToken) {
-      nextPageRequest = Object.assign({}, request);
-      nextPageRequest[this.requestPageTokenField] = pageToken;
-    }
-    const responses = (response as ResponseType)[this.resourceField];
-    this.cache.push(...responses);
-    return nextPageRequest;
-  }
-
-  makeCall(request: RequestType, func: GaxCall, settings: CallSettings) {
-    if (settings.pageToken) {
-      request[this.requestPageTokenField] = settings.pageToken;
-    }
-    if (settings.pageSize) {
-      request[this.requestPageSizeField!] = settings.pageSize;
-    }
-    this.resolveParams([request, func]);
-  }
-
   getApiCaller(settings: CallSettings): APICaller {
     if (!settings.autoPaginate) {
       return new NormalApiCaller();
diff --git a/test/fixtures/google-gax-packaging-test-app/src/index.js b/test/fixtures/google-gax-packaging-test-app/src/index.js
@@ -82,6 +82,7 @@ async function testShowcase() {
   await testEcho(fallbackClient);
   await testPagedExpand(fallbackClient);
   await testWait(fallbackClient);
+  await testPagedExpandAsync(fallbackClient);
 }
 
 async function testEcho(client) {
diff --git a/test/fixtures/google-gax-packaging-test-app/src/v1beta1/echo_client.js b/test/fixtures/google-gax-packaging-test-app/src/v1beta1/echo_client.js
@@ -196,23 +196,23 @@ class EchoClient {
       'wait',
       'pagedExpand',
     ];
-    this._innerCallPromises = {};
     for (const methodName of echoStubMethods) {
-      this._innerCallPromises[methodName] = echoStub.then(
+      const callPromise = echoStub.then(
         stub => (...args) => {
           return stub[methodName].apply(stub, args);
         },
         err => () => {
           throw err;
         }
       );
-      this._innerApiCalls[methodName] = gaxModule.createApiCall(
-        this._innerCallPromises[methodName],
+      const apiCall = gaxModule.createApiCall(
+        callPromise,
         defaults[methodName],
         this._descriptors.page[methodName] ||
           this._descriptors.stream[methodName] ||
           this._descriptors.longrunning[methodName]
       );
+      this._innerApiCalls[methodName] = apiCall;
     }
   }
 
@@ -590,7 +590,7 @@ class EchoClient {
     request = request || {};
     const callSettings = new gax.CallSettings(options);
     return this._descriptors.page.pagedExpand.asyncIterate(
-      this._innerCallPromises['pagedExpand'],
+      this._innerApiCalls.pagedExpand,
       request,
       callSettings
     );
diff --git a/test/unit/pagedIteration.ts b/test/unit/pagedIteration.ts
@@ -22,7 +22,7 @@ import * as sinon from 'sinon';
 import * as streamEvents from 'stream-events';
 import * as through2 from 'through2';
 import {PageDescriptor} from '../../src/paginationCalls/pageDescriptor';
-import {APICallback, GaxCallPromise} from '../../src/apitypes';
+import {APICallback, GaxCall} from '../../src/apitypes';
 import {describe, it, beforeEach} from 'mocha';
 
 import * as util from './utils';
@@ -199,7 +199,7 @@ describe('paged iteration', () => {
 
   describe('use async iterator', () => {
     const spy = sinon.spy(func);
-    let apiCall: GaxCallPromise;
+    let apiCall: GaxCall;
     beforeEach(() => {
       apiCall = util.createApiCall(spy, createOptions);
     });
@@ -210,22 +210,28 @@ describe('paged iteration', () => {
       for await (const resource of iterable) {
         counter++;
         resources.push(resource);
-        if (counter === 10) break;
+        if (counter === 10) {
+          break;
+        }
       }
-      expect(counter).to.equal(10);
+      return resources;
     }
-    it('returns an iterable, count to 10', () => {
+
+    it('returns an iterable, count to 10', async () => {
       const settings = new gax.CallSettings(
         (createOptions && createOptions.settings) || {}
       );
-      iterableChecker(descriptor.asyncIterate(apiCall, {}, settings));
+      const resources = await iterableChecker(
+        descriptor.asyncIterate(apiCall, {}, settings)
+      );
+      expect(resources.length).to.equal(10);
     });
   });
 
   describe('stream conversion', () => {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     let spy: any;
-    let apiCall: GaxCallPromise;
+    let apiCall: GaxCall;
     beforeEach(() => {
       spy = sinon.spy(func);
       apiCall = util.createApiCall(spy, createOptions);

Original file line number	Diff line number	Diff line change
`@@ -82,6 +82,7 @@ async function testShowcase() {`
`82`	`82`	`await testEcho(fallbackClient);`
`83`	`83`	`await testPagedExpand(fallbackClient);`
`84`	`84`	`await testWait(fallbackClient);`
	`85`	`+ await testPagedExpandAsync(fallbackClient);`
`85`	`86`	`}`
`86`	`87`
`87`	`88`	`async function testEcho(client) {`