Move retries createreadstream basic use of api

src/index.ts

@@ -445,6 +445,7 @@ export class Bigtable {
       {},
       baseOptions,
       {
+        gaxServerStreamingRetries: true,
         servicePath: customEndpointBaseUrl || defaultBaseUrl,
         'grpc.callInvocationTransformer': grpcGcp.gcpCallInvocationTransformer,
         'grpc.channelFactoryOverride': grpcGcp.gcpChannelFactoryOverride,

src/table.ts

@@ -14,7 +14,7 @@
 
 import {promisifyAll} from '@google-cloud/promisify';
 import arrify = require('arrify');
-import {ServiceError} from 'google-gax';
+import {GoogleError, ServiceError} from 'google-gax';
 import {BackoffSettings} from 'google-gax/build/src/gax';
 import {PassThrough, Transform} from 'stream';
 
@@ -43,12 +43,13 @@
 import {google} from '../protos/protos';
 import {Duplex} from 'stream';
 import {TableUtils} from './utils/table';
+import * as protos from '../protos/protos';
 
 // See protos/google/rpc/code.proto
 // (4=DEADLINE_EXCEEDED, 8=RESOURCE_EXHAUSTED, 10=ABORTED, 14=UNAVAILABLE)
 const RETRYABLE_STATUS_CODES = new Set([4, 8, 10, 14]);
 // (1=CANCELLED)
 const IGNORED_STATUS_CODES = new Set([1]);
 
 const DEFAULT_BACKOFF_SETTINGS: BackoffSettings = {
   initialRetryDelayMillis: 10,
@@ -719,11 +720,10 @@
    * region_tag:bigtable_api_table_readstream
    */
   createReadStream(opts?: GetRowsOptions) {
-    const options = opts || {};
+    const options: GetRowsOptions = opts || {};
     const maxRetries = is.number(this.maxRetries) ? this.maxRetries! : 10;
     let activeRequestStream: AbortableDuplex | null;
     let rowKeys: string[];
-    let filter: {} | null;
     const rowsLimit = options.limit || 0;
     const hasLimit = rowsLimit !== 0;
     let rowsRead = 0;
@@ -741,10 +741,6 @@
       ranges.push({});
     }
 
-    if (options.filter) {
-      filter = Filter.parse(options.filter);
-    }
-
     let chunkTransformer: ChunkTransformer;
     let rowStream: Duplex;
 
@@ -800,60 +796,26 @@
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
       chunkTransformer = new ChunkTransformer({decode: options.decode} as any);
 
-      const reqOpts = {
-        tableName: this.name,
-        appProfileId: this.bigtable.appProfileId,
-      } as google.bigtable.v2.IReadRowsRequest;
-
+      const shouldRetryFn = function checkRetry(error: GoogleError) {
+        numConsecutiveErrors++;
+        numRequestsMade++;
+        return (
+          numConsecutiveErrors <= maxRetries &&
+          error.code &&
+          (RETRYABLE_STATUS_CODES.has(error.code) || isRstStreamError(error))
+        );
+      };
       const retryOpts = {
         currentRetryAttempt: 0, // was numConsecutiveErrors
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
-        shouldRetryFn: (_: any) => {
-          return false;
-        },
+        shouldRetryFn,
       };
 
       if (lastRowKey) {
-        // Readjust and/or remove ranges based on previous valid row reads.
-        // Iterate backward since items may need to be removed.
-        for (let index = ranges.length - 1; index >= 0; index--) {
-          const range = ranges[index];
-          const startValue = is.object(range.start)
-            ? (range.start as BoundData).value
-            : range.start;
-          const endValue = is.object(range.end)
-            ? (range.end as BoundData).value
-            : range.end;
-          const startKeyIsRead =
-            !startValue ||
-            TableUtils.lessThanOrEqualTo(
-              startValue as string,
-              lastRowKey as string
-            );
-          const endKeyIsNotRead =
-            !endValue ||
-            (endValue as Buffer).length === 0 ||
-            TableUtils.lessThan(lastRowKey as string, endValue as string);
-          if (startKeyIsRead) {
-            if (endKeyIsNotRead) {
-              // EndKey is not read, reset the range to start from lastRowKey open
-              range.start = {
-                value: lastRowKey,
-                inclusive: false,
-              };
-            } else {
-              // EndKey is read, remove this range
-              ranges.splice(index, 1);
-            }
-          }
-        }
-
-        // Remove rowKeys already read.
-        rowKeys = rowKeys.filter(rowKey =>
-          TableUtils.greaterThan(rowKey, lastRowKey as string)
-        );
+        TableUtils.spliceRanges(ranges, lastRowKey);
+        rowKeys = TableUtils.getRowKeys(rowKeys, lastRowKey);
 
         // If there was a row limit in the original request and
         // we've already read all the rows, end the stream and
@@ -870,25 +832,8 @@
         }
       }
 
-      // Create the new reqOpts
-      reqOpts.rows = {};
-
-      // TODO: preprocess all the keys and ranges to Bytes
-      reqOpts.rows.rowKeys = rowKeys.map(
-        Mutation.convertToBytes
-      ) as {} as Uint8Array[];
-
-      reqOpts.rows.rowRanges = ranges.map(range =>
-        Filter.createRange(
-          range.start as BoundData,
-          range.end as BoundData,
-          'Key'
-        )
-      );
-
-      if (filter) {
-        reqOpts.filter = filter;
-      }
+      const reqOpts: protos.google.bigtable.v2.IReadRowsRequest =
+        this.#readRowsReqOpts(ranges, rowKeys, options);
 
       if (hasLimit) {
         reqOpts.rowsLimit = rowsLimit - rowsRead;
@@ -929,7 +874,7 @@
       rowStream = pumpify.obj([requestStream, chunkTransformer, toRowStream]);
 
       // Retry on "received rst stream" errors
-      const isRstStreamError = (error: ServiceError): boolean => {
+      const isRstStreamError = (error: GoogleError): boolean => {
         if (error.code === 13 && error.message) {
           const error_message = (error.message || '').toLowerCase();
           return (
@@ -942,6 +887,7 @@
       };
 
       rowStream
+        /*
         .on('error', (error: ServiceError) => {
           rowStreamUnpipe(rowStream, userStream);
           activeRequestStream = null;
@@ -969,7 +915,8 @@
             userStream.emit('error', error);
           }
         })
+        */
         .on('data', _ => {
           // Reset error count after a successful read so the backoff
           // time won't keep increasing when as stream had multiple errors
           numConsecutiveErrors = 0;
@@ -1592,7 +1539,7 @@
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
         shouldRetryFn: (_: any) => {
           return false;
         },
       };
@@ -1640,6 +1587,40 @@
     makeNextBatchRequest();
   }
 
+  #readRowsReqOpts(
+    ranges: PrefixRange[],
+    rowKeys: string[],
+    options: GetRowsOptions
+  ) {
+    const reqOpts = {
+      tableName: this.name,
+      appProfileId: this.bigtable.appProfileId,
+    } as google.bigtable.v2.IReadRowsRequest;
+
+    // Create the new reqOpts
+    reqOpts.rows = {};
+
+    // TODO: preprocess all the keys and ranges to Bytes
+    reqOpts.rows.rowKeys = rowKeys.map(
+      Mutation.convertToBytes
+    ) as {} as Uint8Array[];
+
+    reqOpts.rows.rowRanges = ranges.map(range =>
+      Filter.createRange(
+        range.start as BoundData,
+        range.end as BoundData,
+        'Key'
+      )
+    );
+
+    const filter = options.filter;
+    if (filter) {
+      reqOpts.filter = Filter.parse(filter);
+    }
+
+    return reqOpts;
+  }
+
   /**
    * Get a reference to a table row.
    *
@@ -2049,7 +2030,7 @@
  * that a callback is omitted.
  */
 promisifyAll(Table, {
-  exclude: ['family', 'row'],
+  exclude: ['family', 'row', '#readRowsReqOpts'],
 });
 
 function getNextDelay(numConsecutiveErrors: number, config: BackoffSettings) {

src/util/mock-servers/mock-server.ts

@@ -37,7 +37,6 @@ export class MockServer {
       `localhost:${this.port}`,
       grpc.ServerCredentials.createInsecure(),
       () => {
-        server.start();
         callback ? callback(portString) : undefined;
       }
     );

src/utils/table.ts

@@ -14,6 +14,8 @@
 
 import {GetRowsOptions, PrefixRange} from '../table';
 import {Mutation} from '../mutation';
+import * as is from 'is';
+import {BoundData} from '../filter';
 
 export class TableUtils {
   static getRanges(options: GetRowsOptions) {
@@ -63,9 +65,53 @@ export class TableUtils {
     return this.lessThan(rhs, lhs);
   }
 
+  static spliceRanges(
+    ranges: PrefixRange[],
+    lastRowKey: string | number | true | Uint8Array
+  ): void {
+    // Readjust and/or remove ranges based on previous valid row reads.
+    // Iterate backward since items may need to be removed.
+    for (let index = ranges.length - 1; index >= 0; index--) {
+      const range = ranges[index];
+      const startValue = is.object(range.start)
+        ? (range.start as BoundData).value
+        : range.start;
+      const endValue = is.object(range.end)
+        ? (range.end as BoundData).value
+        : range.end;
+      const startKeyIsRead =
+        !startValue ||
+        this.lessThanOrEqualTo(startValue as string, lastRowKey as string);
+      const endKeyIsNotRead =
+        !endValue ||
+        (endValue as Buffer).length === 0 ||
+        this.lessThan(lastRowKey as string, endValue as string);
+      if (startKeyIsRead) {
+        if (endKeyIsNotRead) {
+          // EndKey is not read, reset the range to start from lastRowKey open
+          range.start = {
+            value: lastRowKey,
+            inclusive: false,
+          };
+        } else {
+          // EndKey is read, remove this range
+          ranges.splice(index, 1);
+        }
+      }
+    }
+  }
   static lessThanOrEqualTo(lhs: string, rhs: string) {
     return !this.greaterThan(lhs, rhs);
   }
+  static getRowKeys(
+    rowKeys: string[],
+    lastRowKey: string | number | true | Uint8Array
+  ) {
+    // Remove rowKeys already read.
+    return rowKeys.filter(rowKey =>
+      this.greaterThan(rowKey, lastRowKey as string)
+    );
+  }
 
   static createPrefixRange(start: string): PrefixRange {
     const prefix = start.replace(new RegExp('[\xff]+$'), '');

src/v2/bigtable_client.ts

@@ -214,7 +214,7 @@ export class BigtableClient {
       readRows: new this._gaxModule.StreamDescriptor(
         this._gaxModule.StreamType.SERVER_STREAMING,
         !!opts.fallback,
-        /* gaxStreamingRetries: */ false
+        /* gaxStreamingRetries: */ true
       ),
       sampleRowKeys: new this._gaxModule.StreamDescriptor(
         this._gaxModule.StreamType.SERVER_STREAMING,

test/table.ts

@@ -42,7 +42,11 @@ const fakePromisify = Object.assign({}, promisify, {
       return;
     }
     promisified = true;
-    assert.deepStrictEqual(options.exclude, ['family', 'row']);
+    assert.deepStrictEqual(options.exclude, [
+      'family',
+      'row',
+      '#readRowsReqOpts',
+    ]);
   },
 });