chore: remove paginator dependency and use gapic to return streams for paginated calls instead (#762)

Author: AVaksman

handwritten/bigtable/package.json

@@ -47,7 +47,6 @@
     "api-documenter": "api-documenter yaml --input-folder=temp"
   },
   "dependencies": {
-    "@google-cloud/paginator": "^3.0.0",
     "@google-cloud/projectify": "^2.0.0",
     "@google-cloud/promisify": "^2.0.0",
     "arrify": "^2.0.0",

handwritten/bigtable/src/index.ts

@@ -750,16 +750,20 @@ export class Bigtable {
       });
     };
 
+    const gapicStreamingMethods = {listTablesStream: true};
+
     if (isStreamMode) {
       stream = streamEvents(new PassThrough({objectMode: true}));
       stream.abort = () => {
         if (gaxStream && gaxStream.cancel) {
           gaxStream.cancel();
         }
       };
-
-      stream.once('reading', makeRequestStream);
-
+      if (config.method! in gapicStreamingMethods) {
+        stream.once('reading', makeGapicStreamRequest);
+      } else {
+        stream.once('reading', makeRequestStream);
+      }
       return stream;
     } else {
       makeRequestCallback();
@@ -805,6 +809,21 @@ export class Bigtable {
           .pipe(stream);
       });
     }
+
+    function makeGapicStreamRequest() {
+      prepareGaxRequest((err, requestFn) => {
+        if (err) {
+          stream.destroy(err);
+          return;
+        }
+        gaxStream = requestFn!();
+        gaxStream
+          .on('error', (err: Error) => {
+            stream.destroy(err);
+          })
+          .pipe(stream);
+      });
+    }
   }
 
   /**

handwritten/bigtable/src/instance.ts

@@ -12,10 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import {paginator, ResourceStream} from '@google-cloud/paginator';
 import {promisifyAll} from '@google-cloud/promisify';
+import {Transform} from 'stream';
 import arrify = require('arrify');
 import * as is from 'is';
+// eslint-disable-next-line @typescript-eslint/no-var-requires
+const pumpify = require('pumpify');
+
 import snakeCase = require('lodash.snakecase');
 import {
   AppProfile,
@@ -155,7 +158,6 @@ export class Instance {
   id: string;
   name: string;
   metadata?: google.bigtable.admin.v2.IInstance;
-  getTablesStream!: (options?: GetTablesOptions) => ResourceStream<Table>;
   constructor(bigtable: Bigtable, id: string) {
     this.bigtable = bigtable;
 
@@ -905,6 +907,68 @@ Please use the format 'my-instance' or '${bigtable.projectName}/instances/my-ins
     );
   }
 
+  /**
+   * Get {@link Table} objects for all the tables in your Cloud Bigtable
+   * instance as a readable object stream.
+   *
+   * @param {object} [options] Query object. See
+   *     {@link Instance#getTables} for a complete list of options.
+   * @returns {stream}
+   *
+   * @example
+   * const {Bigtable} = require('@google-cloud/bigtable');
+   * const bigtable = new Bigtable();
+   * const instance = bigtable.instance('my-instance');
+   *
+   * instance.getTablesStream()
+   *   .on('error', console.error)
+   *   .on('data', function(table) {
+   *     // table is a Table object.
+   *   })
+   *   .on('end', () => {
+   *     // All tables retrieved.
+   *   });
+   *
+   * //-
+   * // If you anticipate many results, you can end a stream early to prevent
+   * // unnecessary processing and API requests.
+   * //-
+   * instance.getTablesStream()
+   *   .on('data', function(table) {
+   *     this.end();
+   *   });
+   */
+  getTablesStream(options: GetTablesOptions = {}): NodeJS.ReadableStream {
+    const reqOpts = Object.assign({}, options, {
+      parent: this.name,
+      view: Table.VIEWS[options.view || 'unspecified'],
+    });
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    delete (reqOpts as any).gaxOptions;
+
+    // eslint-disable-next-line @typescript-eslint/no-this-alias
+    const self = this;
+    const transformToTable = (
+      chunk: google.bigtable.admin.v2.ITable,
+      enc: string,
+      callback: Function
+    ) => {
+      const table = self.table(chunk.name!.split('/').pop()!);
+      table.metadata = chunk;
+      callback(null, table);
+    };
+    return pumpify.obj([
+      this.bigtable.request({
+        client: 'BigtableTableAdminClient',
+        method: 'listTablesStream',
+        reqOpts,
+        gaxOpts: options.gaxOptions,
+      }),
+      new Transform({objectMode: true, transform: transformToTable}),
+    ]);
+  }
+
   setIamPolicy(
     policy: Policy,
     gaxOptions?: CallOptions
@@ -1104,52 +1168,13 @@ Please use the format 'my-instance' or '${bigtable.projectName}/instances/my-ins
   }
 }
 
-/**
- * Get {@link Table} objects for all the tables in your Cloud Bigtable
- * instance as a readable object stream.
- *
- * @param {object} [query] Configuration object. See
- *     {@link Instance#getTables} for a complete list of options.
- * @returns {stream}
- *
- * @example
- * const {Bigtable} = require('@google-cloud/bigtable');
- * const bigtable = new Bigtable();
- * const instance = bigtable.instance('my-instance');
- *
- * instance.getTablesStream()
- *   .on('error', console.error)
- *   .on('data', function(table) {
- *     // table is a Table object.
- *   })
- *   .on('end', () => {
- *     // All tables retrieved.
- *   });
- *
- * //-
- * // If you anticipate many results, you can end a stream early to prevent
- * // unnecessary processing and API requests.
- * //-
- * instance.getTablesStream()
- *   .on('data', function(table) {
- *     this.end();
- *   });
- */
-Instance.prototype.getTablesStream = paginator.streamify<Table>('getTables');
-
-/*! Developer Documentation
- *
- * These methods can be auto-paginated.
- */
-paginator.extend(Instance, ['getTables']);
-
 /*! Developer Documentation
  *
  * All async methods (except for streams) will return a Promise in the event
  * that a callback is omitted.
  */
 promisifyAll(Instance, {
-  exclude: ['appProfile', 'cluster', 'table'],
+  exclude: ['appProfile', 'cluster', 'table', 'getTablesStream'],
 });
 
 /**

handwritten/bigtable/system-test/bigtable.ts

@@ -160,6 +160,7 @@ describe('Bigtable', () => {
     it('should retrieve a list of app profiles', async () => {
       const [appProfiles] = await INSTANCE.getAppProfiles();
       assert(appProfiles[0] instanceof AppProfile);
+      assert(appProfiles.length > 0);
     });
 
     it('should check if an app profile exists', async () => {

handwritten/bigtable/test/index.ts

@@ -873,6 +873,80 @@ describe('Bigtable', () => {
         requestStream.on('request', done);
       });
     });
+
+    describe('makeGapicStreamRequest', () => {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      let GAX_STREAM: any;
+      const config = {
+        client: 'client',
+        method: 'listTablesStream',
+        reqOpts: {
+          a: 'b',
+          c: 'd',
+        },
+        gaxOpts: {},
+      };
+
+      beforeEach(() => {
+        GAX_STREAM = new PassThrough();
+        bigtable.api[config.client][config.method] = {
+          bind() {
+            return () => {
+              return GAX_STREAM;
+            };
+          },
+        };
+      });
+
+      it('should expose an abort function', done => {
+        GAX_STREAM.cancel = done;
+
+        const requestStream = bigtable.request(config);
+        requestStream.emit('reading');
+        requestStream.abort();
+      });
+
+      it('should prepare the request once reading', done => {
+        bigtable.api[config.client][config.method] = {
+          bind(gaxClient: {}, reqOpts: {}, gaxOpts: {}) {
+            assert.strictEqual(gaxClient, bigtable.api[config.client]);
+            assert.deepStrictEqual(reqOpts, config.reqOpts);
+            assert.strictEqual(gaxOpts, config.gaxOpts);
+            setImmediate(done);
+            return () => {
+              return GAX_STREAM;
+            };
+          },
+        };
+
+        const requestStream = bigtable.request(config);
+        requestStream.emit('reading');
+      });
+
+      it('should destroy the stream with prepare error', done => {
+        const error = new Error('Error.');
+        bigtable.getProjectId_ = (callback: Function) => {
+          callback(error);
+        };
+        const requestStream = bigtable.request(config);
+        requestStream.emit('reading');
+        requestStream.on('error', (err: Error) => {
+          assert.strictEqual(err, error);
+          done();
+        });
+      });
+
+      it('should destroy the stream with GAX error', done => {
+        const error = new Error('Error.');
+        const requestStream = bigtable.request(config);
+        requestStream.emit('reading');
+        GAX_STREAM.emit('error', error);
+        requestStream.on('error', (err: Error) => {
+          assert.strictEqual(err, error);
+          done();
+        });
+      });
+    });
   });
 
   describe('getProjectId_', () => {