refactor: index.ts to async-first

src/index.ts

@@ -17,7 +17,7 @@
 import * as common from '@google-cloud/common-grpc';
 import {paginator} from '@google-cloud/paginator';
 import {replaceProjectIdToken} from '@google-cloud/projectify';
-import {promisifyAll} from '@google-cloud/promisify';
+import {callbackifyAll, promisifyAll} from '@google-cloud/promisify';
 import * as arrify from 'arrify';
 import * as extend from 'extend';
 import {GoogleAuth} from 'google-auth-library';
@@ -224,6 +224,8 @@ class Logging {
   options: LoggingOptions;
   projectId: string;
   detectedResource?: object;
+  configService?: typeof v2.ConfigServiceV2Client;
+  loggingService?: typeof v2.LoggingServiceV2Client;
 
   constructor(options?: LoggingOptions) {
     // Determine what scopes are needed.
@@ -252,6 +254,8 @@ class Logging {
     this.auth = new GoogleAuth(options_);
     this.options = options_;
     this.projectId = this.options.projectId || '{{projectId}}';
+    this.configService = new v2.ConfigServiceV2Client(this.options);
+    this.loggingService = new v2.LoggingServiceV2Client(this.options);
   }
   /**
    * Config to set for the sink. Not all available options are listed here, see
@@ -336,49 +340,34 @@ class Logging {
   createSink(
       name: string, config: CreateSinkRequest,
       callback: CreateSinkCallback): void;
-  createSink(
-      name: string, config: CreateSinkRequest,
-      callback?: CreateSinkCallback): Promise<[Sink, LogSink]>|void {
-    const self = this;
-    if (!is.string(name)) {
+  async createSink(name: string, config: CreateSinkRequest):
+      Promise<[Sink, LogSink]> {
+    if (typeof name !== 'string') {
       throw new Error('A sink name must be provided.');
     }
-    if (!is.object(config)) {
+    if (typeof config !== 'object') {
       throw new Error('A sink configuration object must be provided.');
     }
     if (common.util.isCustomType(config.destination, 'bigquery/dataset')) {
-      this.setAclForDataset_(name, config, callback!);
-      return;
+      await this.setAclForDataset_(config);
     }
     if (common.util.isCustomType(config.destination, 'pubsub/topic')) {
-      this.setAclForTopic_(name, config, callback!);
-      return;
+      await this.setAclForTopic_(config);
     }
     if (common.util.isCustomType(config.destination, 'storage/bucket')) {
-      this.setAclForBucket_(name, config, callback!);
-      return;
+      await this.setAclForBucket_(config);
     }
     const reqOpts = {
       parent: 'projects/' + this.projectId,
       sink: extend({}, config, {name}),
     };
     delete reqOpts.sink.gaxOptions;
-    this.request(
-        {
-          client: 'ConfigServiceV2Client',
-          method: 'createSink',
-          reqOpts,
-          gaxOpts: config.gaxOptions,
-        },
-        (err, resp) => {
-          if (err) {
-            callback!(err, null, resp);
-            return;
-          }
-          const sink = self.sink(resp.name);
-          sink.metadata = resp;
-          callback!(null, sink, resp);
-        });
+    await this.setProjectId(reqOpts);
+    const [resp] =
+        await this.configService.createSink(reqOpts, config.gaxOptions);
+    const sink = this.sink(resp.name);
+    sink.metadata = resp;
+    return [sink, resp];
   }
 
   /**
@@ -615,17 +604,42 @@ class Logging {
             autoPaginate: options.autoPaginate,
           },
           options.gaxOptions);
-      requestStream = self.request({
-        client: 'LoggingServiceV2Client',
-        method: 'listLogEntriesStream',
-        reqOpts,
-        gaxOpts: gaxOptions,
+
+      let gaxStream: ClientReadableStream<LogEntry>;
+      requestStream = streamEvents<Duplex>(through.obj());
+      (requestStream as AbortableDuplex).abort = () => {
+        if (gaxStream && gaxStream.cancel) {
+          gaxStream.cancel();
+        }
+      };
+
+      requestStream.once('reading', () => {
+        // tslint:disable-next-line no-any
+        if ((global as any).GCLOUD_SANDBOX_ENV) {
+          return through.obj();
+        }
+        self.setProjectId(reqOpts).then(() => {
+          try {
+            gaxStream =
+                this.loggingService.listLogEntriesStream(reqOpts, gaxOptions);
+          } catch (error) {
+            requestStream.destroy(error);
+          }
+          gaxStream
+              .on('error',
+                  err => {
+                    requestStream.destroy(err);
+                  })
+              .pipe(requestStream);
+        });
+        return;
       });
       // tslint:disable-next-line no-any
       (userStream as any).setPipeline(requestStream, toEntryStream);
     });
     return userStream;
   }
+
   /**
    * Query object for listing sinks.
    *
@@ -756,7 +770,7 @@ class Logging {
     const self = this;
     options = options || {};
     let requestStream: Duplex;
-    const userStream = streamEvents(pumpify.obj());
+    const userStream = streamEvents<Duplex>(pumpify.obj());
     (userStream as AbortableDuplex).abort = () => {
       if (requestStream) {
         (requestStream as AbortableDuplex).abort();
@@ -777,11 +791,33 @@ class Logging {
             autoPaginate: options.autoPaginate,
           },
           options.gaxOptions);
-      requestStream = self.request({
-        client: 'ConfigServiceV2Client',
-        method: 'listSinksStream',
-        reqOpts,
-        gaxOpts: gaxOptions,
+
+      let gaxStream: ClientReadableStream<LogSink>;
+      requestStream = streamEvents<Duplex>(through.obj());
+      (requestStream as AbortableDuplex).abort = () => {
+        if (gaxStream && gaxStream.cancel) {
+          gaxStream.cancel();
+        }
+      };
+      requestStream.once('reading', () => {
+        // tslint:disable-next-line no-any
+        if ((global as any).GCLOUD_SANDBOX_ENV) {
+          return through.obj();
+        }
+        self.setProjectId(reqOpts).then(() => {
+          try {
+            gaxStream = this.configService.listSinksStream(reqOpts, gaxOptions);
+          } catch (error) {
+            requestStream.destroy(error);
+          }
+          gaxStream
+              .on('error',
+                  err => {
+                    requestStream.destroy(err);
+                  })
+              .pipe(requestStream);
+        });
+        return;
       });
       // tslint:disable-next-line no-any
       (userStream as any).setPipeline(requestStream, toSinkStream);
@@ -919,22 +955,11 @@ class Logging {
    *
    * @private
    */
-  setAclForBucket_(
-      name: string, config: CreateSinkRequest, callback: CreateSinkCallback) {
-    const self = this;
+  async setAclForBucket_(config: CreateSinkRequest) {
     const bucket = config.destination as Bucket;
     // tslint:disable-next-line no-any
-    (bucket.acl.owners as any)
-        .addGroup(
-            '[email protected]',
-            (err: Error, apiResp: request.Response) => {
-              if (err) {
-                callback(err, null, apiResp);
-                return;
-              }
-              config.destination = 'storage.googleapis.com/' + bucket.name;
-              self.createSink(name, config, callback);
-            });
+    await (bucket.acl.owners as any).addGroup('[email protected]');
+    config.destination = 'storage.googleapis.com/' + bucket.name;
   }
 
   /**
@@ -946,37 +971,22 @@ class Logging {
    *
    * @private
    */
-  setAclForDataset_(
-      name: string, config: CreateSinkRequest, callback: CreateSinkCallback) {
-    const self = this;
+  async setAclForDataset_(config: CreateSinkRequest) {
     const dataset = config.destination as Dataset;
-    dataset.getMetadata((err, metadata, apiResp) => {
-      if (err) {
-        callback(err, null, apiResp);
-        return;
-      }
-      // tslint:disable-next-line no-any
-      const access = ([] as any[]).slice.call(arrify(metadata.access));
-      access.push({
-        role: 'WRITER',
-        groupByEmail: '[email protected]',
-      });
-      dataset.setMetadata(
-          {
-            access,
-          },
-          (err, apiResp) => {
-            if (err) {
-              callback(err, null, apiResp);
-              return;
-            }
-            const baseUrl = 'bigquery.googleapis.com';
-            const pId = (dataset.parent as BigQuery).projectId;
-            const dId = dataset.id;
-            config.destination = `${baseUrl}/projects/${pId}/datasets/${dId}`;
-            self.createSink(name, config, callback);
-          });
+    const [metadata] = await dataset.getMetadata();
+    // tslint:disable-next-line no-any
+    const access = ([] as any[]).slice.call(arrify(metadata.access));
+    access.push({
+      role: 'WRITER',
+      groupByEmail: '[email protected]',
+    });
+    await dataset.setMetadata({
+      access,
     });
+    const baseUrl = 'bigquery.googleapis.com';
+    const pId = (dataset.parent as BigQuery).projectId;
+    const dId = dataset.id;
+    config.destination = `${baseUrl}/projects/${pId}/datasets/${dId}`;
   }
 
   /**
@@ -988,34 +998,36 @@ class Logging {
    *
    * @private
    */
-  setAclForTopic_(
-      name: string, config: CreateSinkRequest, callback: CreateSinkCallback) {
-    const self = this;
+  async setAclForTopic_(config: CreateSinkRequest) {
     const topic = config.destination as Topic;
-    topic.iam.getPolicy((err, policy) => {
-      if (err) {
-        callback(err, null);
-        return;
-      }
-      policy!.bindings = arrify(policy!.bindings);
-      policy!.bindings.push({
-        role: 'roles/pubsub.publisher',
-        members: ['serviceAccount:[email protected]'],
-      });
-      topic.iam.setPolicy(policy!, (err, policy) => {
-        if (err) {
-          callback(err, null);
-          return;
-        }
-        const baseUrl = 'pubsub.googleapis.com';
-        const topicName = topic.name;
-        config.destination = `${baseUrl}/${topicName}`;
-        self.createSink(name, config, callback);
-      });
+    const [policy] = await topic.iam.getPolicy();
+    policy.bindings = arrify(policy.bindings);
+    policy!.bindings.push({
+      role: 'roles/pubsub.publisher',
+      members: ['serviceAccount:[email protected]'],
     });
+    await topic.iam.setPolicy(policy);
+    const baseUrl = 'pubsub.googleapis.com';
+    const topicName = topic.name;
+    config.destination = `${baseUrl}/${topicName}`;
+  }
+
+  async setProjectId(reqOpts: {}) {
+    if (this.projectId === '{{projectId}}') {
+      this.projectId = await this.auth.getProjectId();
+    }
+    reqOpts = replaceProjectIdToken(reqOpts, this.projectId!);
   }
 }
 
+/*! Developer Documentation
+ * All async methods (except for streams) will execute a callback in the event
+ * that a callback is provided.
+ */
+callbackifyAll(Logging, {
+  exclude: ['request', 'getEntries', 'getSinks'],
+});
+
 /*! Developer Documentation
  *
  * These methods can be auto-paginated.
@@ -1028,7 +1040,7 @@ paginator.extend(Logging, ['getEntries', 'getSinks']);
  * that a callback is omitted.
  */
 promisifyAll(Logging, {
-  exclude: ['entry', 'log', 'request', 'sink'],
+  exclude: ['entry', 'log', 'request', 'sink', 'createSink'],
 });
 
 /**