fix: return API objects in streams

package.json

@@ -66,6 +66,7 @@
     "merge-stream": "^2.0.0",
     "p-queue": "^6.0.2",
     "protobufjs": "^6.8.6",
+    "pumpify": "2.0.1",
     "split-array-stream": "^2.0.0",
     "stack-trace": "0.0.10",
     "stream-events": "^1.0.4",

src/database.ts

@@ -32,6 +32,7 @@ import {grpc, Operation as GaxOperation, CallOptions} from 'google-gax';
 import {Backup} from './backup';
 import {BatchTransaction, TransactionIdentifier} from './batch-transaction';
 import {google as databaseAdmin} from '../protos/protos';
+import * as pumpify from 'pumpify';
 import {
   Instance,
   CreateDatabaseOptions,
@@ -1348,12 +1349,29 @@ class Database extends GrpcServiceObject {
       delete gaxOpts.pageToken;
     }
 
-    return this.requestStream({
-      client: 'SpannerClient',
-      method: 'listSessionsStream',
-      reqOpts,
-      gaxOpts,
-    });
+    // eslint-disable-next-line @typescript-eslint/no-this-alias
+    const self = this;
+    const transform = function (
+      this: Transform,
+      chunk: databaseAdmin.spanner.v1.ISession,
+      enc: string,
+      callback: Function
+    ) {
+      const session = self.session(chunk.name!);
+      session.metadata = chunk;
+      this.push(session);
+      callback();
+    };
+
+    return new pumpify.obj([
+      this.requestStream({
+        client: 'SpannerClient',
+        method: 'listSessionsStream',
+        reqOpts,
+        gaxOpts,
+      }),
+      new Transform({objectMode: true, transform}),
+    ]);
   }
 
   getSnapshot(options?: TimestampBounds): Promise<[Snapshot]>;

src/index.ts

@@ -47,6 +47,8 @@ import {PartitionedDml, Snapshot, Transaction} from './transaction';
 import grpcGcpModule = require('grpc-gcp');
 const grpcGcp = grpcGcpModule(grpc);
 import * as v1 from './v1';
+import * as pumpify from 'pumpify';
+import {Transform} from 'stream';
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const gcpApiConfig = require('./spanner_grpc_config.json');
@@ -603,12 +605,29 @@ class Spanner extends GrpcService {
       delete gaxOpts.pageToken;
     }
 
-    return this.requestStream({
-      client: 'InstanceAdminClient',
-      method: 'listInstancesStream',
-      reqOpts,
-      gaxOpts,
-    });
+    // eslint-disable-next-line @typescript-eslint/no-this-alias
+    const self = this;
+    const transform = function (
+      this: Transform,
+      chunk: instanceAdmin.spanner.admin.instance.v1.IInstance,
+      enc: string,
+      callback: Function
+    ) {
+      const instance = self.instance(chunk.name!);
+      instance.metadata = chunk;
+      this.push(instance);
+      callback();
+    };
+
+    return new pumpify.obj([
+      this.requestStream({
+        client: 'InstanceAdminClient',
+        method: 'listInstancesStream',
+        reqOpts,
+        gaxOpts,
+      }),
+      new Transform({objectMode: true, transform}),
+    ]);
   }
 
   getInstanceConfigs(

src/instance.ts

@@ -18,6 +18,7 @@ import arrify = require('arrify');
 import {ServiceObjectConfig, GetConfig} from '@google-cloud/common';
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const common = require('./common-grpc/service-object');
+import * as pumpify from 'pumpify';
 import {promisifyAll} from '@google-cloud/promisify';
 import * as extend from 'extend';
 import snakeCase = require('lodash.snakecase');
@@ -32,7 +33,7 @@ import {
   ResourceCallback,
   PagedOptionsWithFilter,
 } from './common';
-import {Duplex} from 'stream';
+import {Duplex, Transform} from 'stream';
 import {SessionPoolOptions, SessionPool} from './session-pool';
 import {grpc, Operation as GaxOperation} from 'google-gax';
 import {Backup} from './backup';
@@ -420,13 +421,27 @@ class Instance extends common.GrpcServiceObject {
       delete gaxOpts.pageSize;
       delete gaxOpts.pageToken;
     }
+    // eslint-disable-next-line @typescript-eslint/no-this-alias
+    const self = this;
+    const transform = function (
+      this: Transform,
+      chunk: instanceAdmin.spanner.admin.database.v1.IBackup,
+      enc: string,
+      callback: Function
+    ) {
+      this.push(self.backup(chunk.name!));
+      callback();
+    };
 
-    return this.requestStream({
-      client: 'DatabaseAdminClient',
-      method: 'listBackupsStream',
-      reqOpts,
-      gaxOpts,
-    });
+    return new pumpify.obj([
+      this.requestStream({
+        client: 'DatabaseAdminClient',
+        method: 'listBackupsStream',
+        reqOpts,
+        gaxOpts,
+      }),
+      new Transform({objectMode: true, transform}),
+    ]);
   }
 
   getBackupOperations(
@@ -1264,12 +1279,29 @@ class Instance extends common.GrpcServiceObject {
       delete gaxOpts.pageToken;
     }
 
-    return this.requestStream({
-      client: 'DatabaseAdminClient',
-      method: 'listDatabasesStream',
-      reqOpts,
-      gaxOpts,
-    });
+    // eslint-disable-next-line @typescript-eslint/no-this-alias
+    const self = this;
+    const transform = function (
+      this: Transform,
+      chunk: databaseAdmin.spanner.admin.database.v1.IDatabase,
+      enc: string,
+      callback: Function
+    ) {
+      const database = self.database(chunk.name!);
+      database.metadata = chunk;
+      this.push(database);
+      callback();
+    };
+
+    return new pumpify.obj([
+      this.requestStream({
+        client: 'DatabaseAdminClient',
+        method: 'listDatabasesStream',
+        reqOpts,
+        gaxOpts,
+      }),
+      new Transform({objectMode: true, transform}),
+    ]);
   }
 
   getMetadata(

system-test/spanner.ts

@@ -23,7 +23,7 @@ import * as crypto from 'crypto';
 import * as extend from 'extend';
 import * as is from 'is';
 import * as uuid from 'uuid';
-import {Backup, Database, Spanner, Instance} from '../src';
+import {Backup, Database, Instance, Session, Spanner} from '../src';
 import {Key} from '../src/table';
 import {
   ReadRequest,
@@ -94,7 +94,7 @@ describe('Spanner', () => {
   after(async () => {
     if (generateInstanceForTest) {
       // Deleting all backups before an instance can be deleted.
-      await Promise.all(
+      await Promise.all<google.protobuf.IEmpty>(
         RESOURCES_TO_CLEAN.filter(
           resource => resource instanceof Backup
         ).map(backup => backup.delete())
@@ -104,7 +104,7 @@ describe('Spanner', () => {
        * All databasess will automatically be deleted with instance.
        * @see {@link https://cloud.google.com/spanner/docs/reference/rpc/google.spanner.admin.instance.v1#google.spanner.admin.instance.v1.InstanceAdmin.DeleteInstance}
        */
-      await Promise.all(
+      await Promise.all<google.protobuf.IEmpty>(
         RESOURCES_TO_CLEAN.filter(
           resource => resource instanceof Instance
         ).map(instance => instance.delete())
@@ -116,7 +116,7 @@ describe('Spanner', () => {
        * @see {@link https://cloud.google.com/spanner/quotas#administrative_limits}
        */
       const limit = pLimit(5);
-      await Promise.all(
+      await Promise.all<google.protobuf.IEmpty>(
         RESOURCES_TO_CLEAN.map(resource => limit(() => resource.delete()))
       );
     }
@@ -1151,6 +1151,23 @@ describe('Spanner', () => {
       );
     });
 
+    it('should list backups steaming', done => {
+      const backups: Backup[] = [];
+      instance
+        .getBackupsStream()
+        .on('error', assert.ifError)
+        .on('data', backup => {
+          backups.push(backup);
+        })
+        .on('end', () => {
+          assert.ok(backups.length > 0);
+          assert.ok(
+            backups.find(b => b.formattedName_ === backup1.formattedName_)
+          );
+          done();
+        });
+    });
+
     it('should list backups with pagination', async () => {
       const [page1, , resp1] = await instance.getBackups({
         pageSize: 1,
@@ -1362,6 +1379,26 @@ describe('Spanner', () => {
 
       await Promise.all(sessions.map(session => session.delete()));
     });
+
+    it('should list sessions', async () => {
+      const [sessions] = await DATABASE.getSessions();
+      assert.ok(sessions.length > 0);
+      assert.ok(sessions.find(s => s.id === session.id));
+    });
+
+    it('should list sessions streaming', done => {
+      const sessions: Session[] = [];
+      DATABASE.getSessionsStream()
+        .on('error', assert.ifError)
+        .on('data', sessionObj => {
+          sessions.push(sessionObj);
+        })
+        .on('end', () => {
+          assert.ok(sessions.length > 0);
+          assert.ok(sessions.find(s => s.id === session.id));
+          done();
+        });
+    });
   });
 
   describe('Tables', () => {

test/database.ts

@@ -23,14 +23,15 @@ import * as extend from 'extend';
 import {ApiError, util} from '@google-cloud/common';
 import * as proxyquire from 'proxyquire';
 import * as sinon from 'sinon';
-import {Transform, Duplex} from 'stream';
+import {Transform} from 'stream';
 import * as through from 'through2';
 import * as pfy from '@google-cloud/promisify';
 import {grpc} from 'google-gax';
 import * as db from '../src/database';
 import {Instance} from '../src';
 import {MockError} from './mockserver/mockspanner';
 import {IOperation} from '../src/instance';
+import duplexify = require('duplexify');
 
 let promisified = false;
 const fakePfy = extend({}, pfy, {
@@ -2020,7 +2021,7 @@ describe('Database', () => {
     const OPTIONS = {
       gaxOptions: {autoPaginate: false},
     } as db.GetSessionsOptions;
-    const returnValue = {} as Duplex;
+    const returnValue = through.obj();
 
     it('should make and return the correct gax API call', () => {
       const expectedReqOpts = extend({}, OPTIONS, {
@@ -2040,7 +2041,7 @@ describe('Database', () => {
       };
 
       const returnedValue = database.getSessionsStream(OPTIONS);
-      assert.strictEqual(returnedValue, returnValue);
+      assert.ok(returnedValue instanceof duplexify);
     });
 
     it('should pass pageSize and pageToken from gaxOptions into reqOpts', () => {
@@ -2067,7 +2068,7 @@ describe('Database', () => {
       };
 
       const returnedValue = database.getSessionsStream(options);
-      assert.strictEqual(returnedValue, returnValue);
+      assert.ok(returnedValue instanceof duplexify);
     });
 
     it('pageSize and pageToken in options should take precedence over gaxOptions', () => {
@@ -2101,7 +2102,7 @@ describe('Database', () => {
       };
 
       const returnedValue = database.getSessionsStream(options);
-      assert.strictEqual(returnedValue, returnValue);
+      assert.ok(returnedValue instanceof duplexify);
     });
 
     it('should not require options', () => {
@@ -2116,7 +2117,29 @@ describe('Database', () => {
       };
 
       const returnedValue = database.getSessionsStream();
-      assert.strictEqual(returnedValue, returnValue);
+      assert.ok(returnedValue instanceof duplexify);
+    });
+
+    it('should create and return Session objects', done => {
+      const stream = through.obj();
+      database.requestStream = () => {
+        return stream;
+      };
+      const protoSession = {name: 'session'};
+      setImmediate(() => {
+        stream.push(protoSession);
+        stream.push(null);
+      });
+
+      database
+        .getSessionsStream()
+        .on('error', assert.ifError)
+        .on('data', session => {
+          assert.ok(session instanceof FakeSession);
+          // eslint-disable-next-line @typescript-eslint/no-explicit-any
+          assert.strictEqual((session as any).metadata, protoSession);
+        })
+        .on('end', done);
     });
   });