fix: system-test issue

Author: Lalji Kanjareeya

src/table.ts

@@ -809,9 +809,13 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
 
       requestStream!.on('request', () => numRequestsMade++);
 
-      const transform = new Transform({
+      const toRowStream = new Transform({
         transform: (rowData, _, next) => {
-          if (chunkTransformer._destroyed || !userStream.writable) {
+          if (
+            chunkTransformer._destroyed ||
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            (userStream as any)._writableState.ended
+          ) {
             return next();
           }
           numRequestsMade = 0;
@@ -823,7 +827,7 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
         objectMode: true,
       });
 
-      rowStream = pumpify.obj([requestStream, chunkTransformer, transform]);
+      rowStream = pumpify.obj([requestStream, chunkTransformer, toRowStream]);
 
       rowStream.on('error', (error: ServiceError) => {
         if (IGNORED_STATUS_CODES.has(error.code)) {
@@ -1557,7 +1561,7 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
       appProfileId: this.bigtable.appProfileId,
     };
 
-    const transform = new Transform({
+    const rowKeysStream = new Transform({
       transform(key, enc, next) {
         next(null, {
           key: key.rowKey,
@@ -1574,7 +1578,7 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
         reqOpts,
         gaxOpts: gaxOptions,
       }),
-      transform,
+      rowKeysStream,
     ]);
   }
 

system-test/read-rows-acceptance-test.json

@@ -378,14 +378,6 @@
         "name": "invalid - duplicate row key",
         "chunks_base64": ["CgRSS18xEgMKAUEaAwoBQyBkMgl2YWx1ZS1WQUxIAQ==", "CgRSS18xEgMKAUIaAwoBRCBkMgl2YWx1ZS1WQUxIAQ=="],
         "results": [{
-            "fm": "A",
-            "rk": "RK_1",
-            "qual": "C",
-            "label": "",
-            "error": false,
-            "value": "value-VAL",
-            "ts": 100
-        }, {
             "fm": "",
             "rk": "",
             "qual": "",
@@ -399,14 +391,6 @@
         "name": "invalid - new row missing row key",
         "chunks_base64": ["CgRSS18xEgMKAUEaAwoBQyBkMgl2YWx1ZS1WQUxIAQ==", "IGQyCXZhbHVlLVZBTEgB"],
         "results": [{
-            "fm": "A",
-            "rk": "RK_1",
-            "qual": "C",
-            "label": "",
-            "error": false,
-            "value": "value-VAL",
-            "ts": 100
-        }, {
             "fm": "",
             "rk": "",
             "qual": "",

system-test/read-rows-acceptance-tests.ts

@@ -24,9 +24,7 @@ import {protobuf} from 'google-gax';
 import * as fs from 'fs';
 import * as path from 'path';
 import {Instance} from '../src/instance';
-import {Bigtable} from '../src';
-// eslint-disable-next-line @typescript-eslint/no-var-requires
-const streamEvents = require('stream-events');
+import {Bigtable, AbortableDuplex} from '../src';
 
 const protosJson = path.resolve(__dirname, '../protos/protos.json');
 const root = protobuf.Root.fromJSON(
@@ -71,29 +69,31 @@ describe('Read Row Acceptance tests', () => {
       table.bigtable = {} as Bigtable;
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
       (table.bigtable.request as any) = () => {
-        const stream = streamEvents(new PassThrough({objectMode: true}));
-        stream.abort = () => {};
+        const stream = new PassThrough({
+          objectMode: true,
+        });
 
-        const mocked_stream = function (size: number) {
-          test.chunks_base64.forEach((value, index, array) => {
-            const chunk = value;
-            const cellChunk = CellChunk.decode(
-              Buffer.from(chunk as string, 'base64')
-            );
-            // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            let readRowsResponse: any = {chunks: [cellChunk]};
-            readRowsResponse = ReadRowsResponse.create(readRowsResponse);
-            readRowsResponse = ReadRowsResponse.toObject(readRowsResponse, {
-              defaults: true,
-              longs: String,
-              oneofs: true,
-            });
-            stream.emit('data', readRowsResponse);
-          });
-          stream.emit('end');
-        };
+        ((stream as {}) as AbortableDuplex).abort = () => {};
 
-        stream.once('reading', mocked_stream);
+        setImmediate(() => {
+          test.chunks_base64
+            .map(chunk => {
+              const cellChunk = CellChunk.decode(
+                Buffer.from(chunk as string, 'base64')
+              ); //.decode64(chunk);
+              // eslint-disable-next-line @typescript-eslint/no-explicit-any
+              let readRowsResponse: any = {chunks: [cellChunk]};
+              readRowsResponse = ReadRowsResponse.create(readRowsResponse);
+              readRowsResponse = ReadRowsResponse.toObject(readRowsResponse, {
+                defaults: true,
+                longs: String,
+                oneofs: true,
+              });
+              return readRowsResponse;
+            })
+            .forEach(readRowsResponse => stream.push(readRowsResponse));
+          stream.push(null);
+        });
 
         return stream;
       };