fix: also reject Promises on failure, and add tests

googleapis · Jun 16, 2022 · 918a64d · 918a64d
1 parent c9d539b
commit 918a64d
Show file tree

Hide file tree

Showing 2 changed files with 68 additions and 0 deletions.
diff --git a/src/message-queues.ts b/src/message-queues.ts
@@ -237,6 +237,9 @@ export class AckQueue extends MessageQueue {
         responsePromise?.resolve(AckResponses.Success);
       });
     } catch (e) {
+      batch.forEach(({responsePromise}) => {
+        responsePromise?.reject(e);
+      });
       throw new BatchError(e as ServiceError, ackIds, 'acknowledge');
     }
   }
@@ -286,6 +289,9 @@ export class ModAckQueue extends MessageQueue {
           responsePromise?.resolve(AckResponses.Success);
         });
       } catch (e) {
+        batch.forEach(({responsePromise}) => {
+          responsePromise?.reject(e);
+        });
         throw new BatchError(e as ServiceError, ackIds, 'modifyAckDeadline');
       }
     });

diff --git a/test/message-queues.ts b/test/message-queues.ts
@@ -159,6 +159,24 @@ describe('MessageQueues', () => {
         clock.tick(delay);
         assert.strictEqual(stub.callCount, 1);
       });
+
+      it('should return a Promise that resolves when the ack is sent', async () => {
+        const clock = sandbox.useFakeTimers();
+        const delay = 1000;
+        messageQueue.setOptions({maxMilliseconds: delay});
+
+        sandbox
+          .stub(messageQueue, '_sendBatch')
+          .callsFake((batch: messageTypes.QueuedMessages) => {
+            batch.forEach(m => {
+              m.responsePromise?.resolve();
+            });
+          });
+
+        const completion = messageQueue.add(new FakeMessage() as Message);
+        clock.tick(delay);
+        await completion;
+      });
     });
 
     describe('flush', () => {
@@ -391,6 +409,26 @@ describe('MessageQueues', () => {
       messages.forEach(message => ackQueue.add(message as Message));
       ackQueue.flush();
     });
+
+    it('should appropriately resolve result promises', async () => {
+      const stub = sandbox.stub(subscriber.client, 'acknowledge').resolves();
+
+      const message = new FakeMessage() as Message;
+      const completion = ackQueue.add(message);
+      await ackQueue.flush();
+      assert.strictEqual(stub.callCount, 1);
+      await completion;
+    });
+
+    it('should appropriately reject result promises', async () => {
+      const stub = sandbox.stub(subscriber.client, 'acknowledge').resolves();
+
+      const message = new FakeMessage() as Message;
+      const completion = ackQueue.add(message);
+      await ackQueue.flush();
+      assert.strictEqual(stub.callCount, 1);
+      await completion;
+    });
   });
 
   describe('ModAckQueue', () => {
@@ -516,5 +554,29 @@ describe('MessageQueues', () => {
       messages.forEach(message => modAckQueue.add(message as Message));
       modAckQueue.flush();
     });
+
+    it('should appropriately resolve result promises', async () => {
+      const stub = sandbox
+        .stub(subscriber.client, 'modifyAckDeadline')
+        .resolves();
+
+      const message = new FakeMessage() as Message;
+      const completion = modAckQueue.add(message);
+      await modAckQueue.flush();
+      assert.strictEqual(stub.callCount, 1);
+      await completion;
+    });
+
+    it('should appropriately reject result promises', async () => {
+      const stub = sandbox
+        .stub(subscriber.client, 'modifyAckDeadline')
+        .resolves();
+
+      const message = new FakeMessage() as Message;
+      const completion = modAckQueue.add(message);
+      await modAckQueue.flush();
+      assert.strictEqual(stub.callCount, 1);
+      await completion;
+    });
   });
 });