fix: revert some of the promise fixes

Author: feywind

src/message-queues.ts

@@ -200,7 +200,7 @@ export abstract class MessageQueue {
       if (this._subscriber.isExactlyOnceDelivery) {
         throw new AckError(AckResponses.Invalid, 'Subscriber closed');
       } else {
-        return Promise.resolve();
+        return;
       }
     }
 

src/subscription.ts

@@ -589,34 +589,23 @@ 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) {
-      await this._subscriber.close();
+      void this._subscriber.close();
     }
 
-    await promisify(this.request<google.protobuf.Empty>)({
-      client: 'SubscriberClient',
-      method: 'deleteSubscription',
-      reqOpts,
-      gaxOpts,
-    });
-
-    return [{}];
+    this.request<google.protobuf.Empty>(
+      {
+        client: 'SubscriberClient',
+        method: 'deleteSubscription',
+        reqOpts,
+        gaxOpts,
+      },
+      callback!,
+    );
   }
 
   /**

test/message-queues.ts

@@ -146,36 +146,36 @@ describe('MessageQueues', () => {
     });
 
     describe('add', () => {
-      it('should increase the number of pending requests', async () => {
-        await await messageQueue.add(new FakeMessage() as Message);
+      it('should increase the number of pending requests', () => {
+        void messageQueue.add(new FakeMessage() as Message);
         assert.strictEqual(messageQueue.numPendingRequests, 1);
       });
 
-      it('should flush the queue if at capacity', async () => {
+      it('should flush the queue if at capacity', () => {
         const stub = sandbox.stub(messageQueue, 'flush');
 
         messageQueue.setOptions({maxMessages: 1});
-        await await messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.add(new FakeMessage() as Message);
 
         assert.strictEqual(stub.callCount, 1);
       });
 
-      it('should flush the queue if at byte capacity', async () => {
+      it('should flush the queue if at byte capacity', () => {
         const stub = sandbox.stub(messageQueue, 'flush');
 
         messageQueue.bytes = messageTypes.MAX_BATCH_BYTES - 10;
-        await await messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.add(new FakeMessage() as Message);
 
         assert.strictEqual(stub.callCount, 1);
       });
 
-      it('should schedule a flush if needed', async () => {
+      it('should schedule a flush if needed', () => {
         const clock = sandbox.useFakeTimers();
         const stub = sandbox.stub(messageQueue, 'flush');
         const delay = 1000;
 
         messageQueue.setOptions({maxMilliseconds: delay});
-        await await messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.add(new FakeMessage() as Message);
 
         assert.strictEqual(stub.callCount, 0);
         clock.tick(delay);
@@ -196,46 +196,46 @@ describe('MessageQueues', () => {
             return Promise.resolve([]);
           });
 
-        const completion = await messageQueue.add(new FakeMessage() as Message);
+        const completion = messageQueue.add(new FakeMessage() as Message);
         clock.tick(delay);
         await completion;
       });
     });
 
     describe('flush', () => {
-      it('should cancel scheduled flushes', async () => {
+      it('should cancel scheduled flushes', () => {
         const clock = sandbox.useFakeTimers();
         const spy = sandbox.spy(messageQueue, 'flush');
         const delay = 1000;
 
         messageQueue.setOptions({maxMilliseconds: delay});
-        await await messageQueue.add(new FakeMessage() as Message);
-        await await messageQueue.flush();
+        void messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.flush();
         clock.tick(delay);
 
         assert.strictEqual(spy.callCount, 1);
       });
 
-      it('should remove the messages from the queue', async () => {
-        await await messageQueue.add(new FakeMessage() as Message);
-        await await messageQueue.flush();
+      it('should remove the messages from the queue', () => {
+        void messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.flush();
 
         assert.strictEqual(messageQueue.numPendingRequests, 0);
       });
 
-      it('should remove the bytes of messages from the queue', async () => {
-        await await messageQueue.add(new FakeMessage() as Message);
-        await await messageQueue.flush();
+      it('should remove the bytes of messages from the queue', () => {
+        void messageQueue.add(new FakeMessage() as Message);
+        void messageQueue.flush();
 
         assert.strictEqual(messageQueue.bytes, 0);
       });
 
-      it('should send the batch', async () => {
+      it('should send the batch', () => {
         const message = new FakeMessage();
         const deadline = 10;
 
-        await messageQueue.add(message as Message, deadline);
-        await messageQueue.flush();
+        void messageQueue.add(message as Message, deadline);
+        void messageQueue.flush();
 
         const [batch] = messageQueue.batches;
         assert.strictEqual(batch[0].message.ackId, message.ackId);
@@ -277,8 +277,8 @@ describe('MessageQueues', () => {
         const onDrainBeforeFlush = messageQueue
           .onDrain()
           .then(() => log.push('drain1'));
-        await messageQueue.add(message as Message, deadline);
-        await messageQueue.flush();
+        void messageQueue.add(message as Message, deadline);
+        void messageQueue.flush();
         assert.deepStrictEqual(log, ['send:start']);
         sendDone.resolve();
         await messageQueue.onDrain().then(() => log.push('drain2'));
@@ -324,48 +324,48 @@ describe('MessageQueues', () => {
     });
 
     describe('setOptions', () => {
-      it('should default maxMessages to 3000', async () => {
+      it('should default maxMessages to 3000', () => {
         const stub = sandbox.stub(messageQueue, 'flush');
 
         for (let i = 0; i < 3000; i++) {
           assert.strictEqual(stub.callCount, 0);
-          await messageQueue.add(fakeMessage());
+          void messageQueue.add(fakeMessage());
         }
 
         assert.strictEqual(stub.callCount, 1);
       });
 
-      it('should respect user supplied maxMessages', async () => {
+      it('should respect user supplied maxMessages', () => {
         const stub = sandbox.stub(messageQueue, 'flush');
         const maxMessages = 100;
 
         messageQueue.setOptions({maxMessages});
 
         for (let i = 0; i < maxMessages; i++) {
           assert.strictEqual(stub.callCount, 0);
-          await messageQueue.add(fakeMessage());
+          void messageQueue.add(fakeMessage());
         }
 
         assert.strictEqual(stub.callCount, 1);
       });
 
-      it('should default maxMilliseconds to 100', async () => {
+      it('should default maxMilliseconds to 100', () => {
         const clock = sandbox.useFakeTimers();
         const stub = sandbox.stub(messageQueue, 'flush');
 
-        await messageQueue.add(fakeMessage());
+        void messageQueue.add(fakeMessage());
         clock.tick(100);
 
         assert.strictEqual(stub.callCount, 1);
       });
 
-      it('should respect user supplied maxMilliseconds', async () => {
+      it('should respect user supplied maxMilliseconds', () => {
         const clock = sandbox.useFakeTimers();
         const stub = sandbox.stub(messageQueue, 'flush');
         const maxMilliseconds = 10000;
 
         messageQueue.setOptions({maxMilliseconds});
-        await messageQueue.add(fakeMessage());
+        void messageQueue.add(fakeMessage());
         clock.tick(maxMilliseconds);
 
         assert.strictEqual(stub.callCount, 1);
@@ -569,7 +569,7 @@ describe('MessageQueues', () => {
         };
 
         sandbox.stub(fakeSubscriber.client, 'acknowledge').rejects(fakeError);
-        await ackQueue.add(message);
+        void ackQueue.add(message);
         await ackQueue.flush();
 
         // Make sure the one handled by errorInfo was retried.

test/publisher/index.ts

@@ -193,13 +193,13 @@ describe('Publisher', () => {
       exporter.reset();
     });
 
-    it('export created spans', async () => {
+    it('export created spans', () => {
       tracing.setGloballyEnabled(true);
 
       // Setup trace exporting
       tracingPublisher = new Publisher(topic);
       const msg = {data: buffer} as p.PubsubMessage;
-      await tracingPublisher.publishMessage(msg);
+      void tracingPublisher.publishMessage(msg);
 
       // publishMessage is only the first part of the process now,
       // so we need to manually end the span.

test/publisher/message-queues.ts

@@ -155,23 +155,23 @@ describe('Message Queues', () => {
       const messages = [{}, {}, {}];
       const callbacks = messages.map(() => sandbox.spy());
 
-      it('should make the correct request', async () => {
+      it('should make the correct request', () => {
         const stub = sandbox.stub(topic, 'request');
 
-        await queue._publish(messages, callbacks);
+        void queue._publish(messages, callbacks);
 
         const [{client, method, reqOpts}] = stub.lastCall.args;
         assert.strictEqual(client, 'PublisherClient');
         assert.strictEqual(method, 'publish');
         assert.deepStrictEqual(reqOpts, {topic: topic.name, messages});
       });
 
-      it('should pass along any gax options', async () => {
+      it('should pass along any gax options', () => {
         const stub = sandbox.stub(topic, 'request');
         const callOptions = {};
 
         publisher.settings.gaxOpts = callOptions;
-        await queue._publish(messages, callbacks);
+        void queue._publish(messages, callbacks);
 
         const [{gaxOpts}] = stub.lastCall.args;
         assert.strictEqual(gaxOpts, callOptions);
@@ -659,27 +659,27 @@ describe('Message Queues', () => {
         queue.currentBatch.callbacks = spies;
       });
 
-      it('should set inFlight to true', async () => {
-        await queue.publish();
+      it('should set inFlight to true', () => {
+        void queue.publish();
         assert.strictEqual(queue.inFlight, true);
       });
 
-      it('should cancel any pending publishes', async () => {
+      it('should cancel any pending publishes', () => {
         const fakeHandle = 1234 as unknown as NodeJS.Timeout;
         const stub = sandbox.stub(global, 'clearTimeout');
 
         queue.pending = fakeHandle;
-        await queue.publish();
+        void queue.publish();
 
         const [handle] = stub.lastCall.args;
         assert.strictEqual(handle, fakeHandle);
         assert.strictEqual(queue.pending, undefined);
       });
 
-      it('should remove the oldest batch from the batch list', async () => {
+      it('should remove the oldest batch from the batch list', () => {
         const oldestBatch = queue.currentBatch;
 
-        await queue.publish();
+        void queue.publish();
 
         assert.notStrictEqual(queue.currentBatch, oldestBatch);
       });

test/subscriber.ts

@@ -492,7 +492,7 @@ describe('Subscriber', () => {
       assert.strictEqual(subscriber.ackDeadline, 10);
     });
 
-    it('should not update the deadline if user specified', async () => {
+    it('should not update the deadline if user specified', () => {
       const histogram: FakeHistogram = stubs.get('histogram');
       const ackDeadline = 543;
       const maxMessages = 20;
@@ -505,16 +505,16 @@ describe('Subscriber', () => {
         ackDeadline,
         flowControl: {maxMessages: maxMessages, maxBytes: maxBytes},
       });
-      await subscriber.ack(message);
+      void subscriber.ack(message);
 
       assert.strictEqual(subscriber.ackDeadline, ackDeadline);
     });
 
-    it('should add the message to the ack queue', async () => {
+    it('should add the message to the ack queue', () => {
       const ackQueue: FakeAckQueue = stubs.get('ackQueue');
       const stub = sandbox.stub(ackQueue, 'add').withArgs(message);
 
-      await subscriber.ack(message);
+      void subscriber.ack(message);
 
       assert.strictEqual(stub.callCount, 1);
     });
@@ -662,11 +662,11 @@ describe('Subscriber', () => {
   describe('modAck', () => {
     const deadline = 600;
 
-    it('should add the message/deadline to the modAck queue', async () => {
+    it('should add the message/deadline to the modAck queue', () => {
       const modAckQueue: FakeModAckQueue = stubs.get('modAckQueue');
       const stub = sandbox.stub(modAckQueue, 'add').withArgs(message, deadline);
 
-      await subscriber.modAck(message, deadline);
+      void subscriber.modAck(message, deadline);
 
       assert.strictEqual(stub.callCount, 1);
     });