feat: fix a bunch of linter errors, and import new gapic classes

Author: feywind

src/message-queues.ts

@@ -195,7 +195,7 @@ export abstract class MessageQueue {
    * @param {number} [deadline] The deadline in seconds.
    * @private
    */
-  add(message: Message, deadline?: number): Promise<void> {
+  async add(message: Message, deadline?: number): Promise<void> {
     if (this._closed) {
       if (this._subscriber.isExactlyOnceDelivery) {
         throw new AckError(AckResponses.Invalid, 'Subscriber closed');
@@ -213,7 +213,7 @@ export abstract class MessageQueue {
       this._requests.length + 1 >= maxMessages! ||
       this.bytes + size >= MAX_BATCH_BYTES
     ) {
-      this.flush();
+      await this.flush();
     }
 
     // Add the message to the current batch.

src/message-stream.ts

@@ -346,7 +346,7 @@ export class MessageStream extends PassThrough {
    */
   private async _getClient(): Promise<ClientStub> {
     const client = await this._subscriber.getClient();
-    client.initialize();
+    await client.initialize();
     return client.subscriberStub as Promise<ClientStub>;
   }
 
@@ -397,9 +397,9 @@ export class MessageStream extends PassThrough {
       if (PullRetry.resetFailures(status)) {
         this._retrier.reset(this._streams[index]);
       }
-      this._retrier.retryLater(this._streams[index], () => {
-        this._fillOne(index);
-      });
+      this._retrier.retryLater(this._streams[index], () =>
+        this._fillOne(index),
+      );
     } else if (this._activeStreams() === 0) {
       this.emit(
         'debug',

src/publisher/flow-publisher.ts

@@ -90,7 +90,7 @@ export class FlowControlledPublisher {
       const waitPromise = this.flowControl.willSend(size, 1);
       return waitPromise.then(doPublish);
     } else {
-      this.flowControl.willSend(size, 1).then(() => {});
+      void this.flowControl.willSend(size, 1).then(() => {});
       doPublish();
       return null;
     }

src/subscriber.ts

@@ -406,7 +406,7 @@ export class Message implements tracing.MessageWithAttributes {
       this._handled = true;
       this.subSpans.ackCall();
       this.subSpans.processingEnd();
-      this._subscriber.ack(this);
+      void this._subscriber.ack(this);
     }
   }
 
@@ -456,7 +456,7 @@ export class Message implements tracing.MessageWithAttributes {
   modAck(deadline: number): void {
     if (!this._handled) {
       this.subSpans.modAckCall(Duration.from({seconds: deadline}));
-      this._subscriber.modAck(this, deadline);
+      void this._subscriber.modAck(this, deadline);
     }
   }
 
@@ -506,7 +506,7 @@ export class Message implements tracing.MessageWithAttributes {
       this._handled = true;
       this.subSpans.nackCall();
       this.subSpans.processingEnd();
-      this._subscriber.nack(this);
+      void this._subscriber.nack(this);
     }
   }
 
@@ -951,7 +951,7 @@ export class Subscriber extends EventEmitter {
 
     this._stream.start().catch(err => {
       this.emit('error', err);
-      this.close();
+      void this.close();
     });
 
     this.isOpen = true;
@@ -1106,12 +1106,12 @@ export class Subscriber extends EventEmitter {
 
     if (this._acks.numPendingRequests) {
       promises.push(this._acks.onFlush());
-      this._acks.flush();
+      await this._acks.flush();
     }
 
     if (this._modAcks.numPendingRequests) {
       promises.push(this._modAcks.onFlush());
-      this._modAcks.flush();
+      await this._modAcks.flush();
     }
 
     if (this._acks.numInFlightRequests) {

src/subscription.ts

@@ -47,6 +47,7 @@ import {promisifySome} from './util';
 import {StatusError} from './message-stream';
 import {DebugMessage} from './debug';
 import {EventEmitter} from 'stream';
+import {promisify} from 'util';
 
 export {AckError, AckResponse, AckResponses} from './subscriber';
 
@@ -588,23 +589,34 @@ export class Subscription extends EventEmitter {
     const gaxOpts = typeof optsOrCallback === 'object' ? optsOrCallback : {};
     callback = typeof optsOrCallback === 'function' ? optsOrCallback : callback;
 
+    const prom = this._deleteAsync(gaxOpts);
+    if (callback) {
+      prom.then(r => callback(null, r)).catch(e => callback(e));
+    } else {
+      return prom;
+    }
+  }
+
+  /*!
+   * Async innards of delete() so we can wait for subscription.close() if needed.
+   */
+  async _deleteAsync(gaxOpts: CallOptions): Promise<EmptyResponse> {
     const reqOpts = {
       subscription: this.name,
     };
 
     if (this.isOpen) {
-      this._subscriber.close();
+      await this._subscriber.close();
     }
 
-    this.request<google.protobuf.Empty>(
-      {
-        client: 'SubscriberClient',
-        method: 'deleteSubscription',
-        reqOpts,
-        gaxOpts,
-      },
-      callback!,
-    );
+    await promisify(this.request<google.protobuf.Empty>)({
+      client: 'SubscriberClient',
+      method: 'deleteSubscription',
+      reqOpts,
+      gaxOpts,
+    });
+
+    return [{}];
   }
 
   /**
@@ -1164,7 +1176,7 @@ export class Subscription extends EventEmitter {
 
     this.on('removeListener', () => {
       if (this.isOpen && this.listenerCount('message') === 0) {
-        this._subscriber.close();
+        void this._subscriber.close();
       }
     });
   }