fix(WaitForProcessing): Shutdown option WaitForProcessing never entered on subscription close

src/lease-manager.ts

@@ -106,6 +106,8 @@ export class LeaseManager extends EventEmitter {
    * Adds a message to the inventory, kicking off the deadline extender if it
    * isn't already running.
    *
+   * @fires LeaseManager#full
+   *
    * @param {Message} message The message.
    * @private
    */
@@ -141,6 +143,10 @@ export class LeaseManager extends EventEmitter {
   }
   /**
    * Removes ALL messages from inventory, and returns the ones removed.
+   *
+   * @fires LeaseManager#free
+   * @fires LeaseManager#empty
+   *
    * @private
    */
   clear(): Message[] {
@@ -197,6 +203,7 @@ export class LeaseManager extends EventEmitter {
    * messages are left over.
    *
    * @fires LeaseManager#free
+   * @fires LeaseManager#empty
    *
    * @param {Message} message The message to remove.
    * @private

src/subscriber.ts

@@ -1000,7 +1000,7 @@ export class Subscriber extends EventEmitter {
     const shutdownStart = Date.now();
     if (
       behavior === SubscriberCloseBehaviors.WaitForProcessing &&
-      !this._inventory.isEmpty
+      !this._inventory.isEmpty()
     ) {
       const waitTimeout = timeout.subtract(FINAL_NACK_TIMEOUT);
 

system-test/pubsub.ts

@@ -429,8 +429,8 @@ describe('pubsub', () => {
     const SUB_NAMES = [generateSubName(), generateSubName()];
     const SUB_DETACH_NAME = generateSubForDetach();
 
-    const thirty = Duration.from({minutes: 30});
-    const sixty = Duration.from({minutes: 60});
+    const thirty = Duration.from({seconds: 30});
+    const sixty = Duration.from({seconds: 60});
     const SUBSCRIPTIONS = [
       topic.subscription(SUB_NAMES[0], {minAckDeadline: thirty, maxAckDeadline: thirty}),
       topic.subscription(SUB_NAMES[1], {minAckDeadline: sixty, maxAckDeadline: sixty}),
@@ -660,8 +660,9 @@ describe('pubsub', () => {
       subscription.on('error', done);
 
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      subscription.on('message', (message: {data: any}) => {
+      subscription.on('message', message => {
         assert.deepStrictEqual(message.data, Buffer.from('hello'));
+        message.ack();
 
         if (++messageCount === 10) {
           subscription.close(done);

test/subscriber.ts

@@ -106,7 +106,7 @@ class FakeLeaseManager extends EventEmitter {
   remove(message: s.Message): void {}
 
   _isEmpty = true;
-  get isEmpty() {
+  isEmpty(): boolean {
     return this._isEmpty;
   }
 }