feat: Add optional delay when calling nack() (#255) (#256)

googleapis · Nov 19, 2018 · b35005e · b35005e
1 parent 6e420e7
commit b35005e
Show file tree

Hide file tree

Showing 4 changed files with 92 additions and 13 deletions.
diff --git a/src/connection-pool.ts b/src/connection-pool.ts
@@ -301,8 +301,8 @@ export class ConnectionPool extends EventEmitter {
       ack: () => {
         this.subscription.ack_(message);
       },
-      nack: () => {
-        this.subscription.nack_(message);
+      nack: (delay?: number) => {
+        this.subscription.nack_(message, delay);
       }
     }
     return message;

diff --git a/src/subscriber.ts b/src/subscriber.ts
@@ -237,24 +237,41 @@ export class Subscriber extends EventEmitter {
     }
     const acks = this.inventory_.ack;
     const nacks = this.inventory_.nack;
+
     if (!acks.length && !nacks.length) {
       return Promise.resolve();
     }
+
     const requests: Promise<void>[] = [];
+
     if (acks.length) {
       requests.push(
         this.acknowledge_(acks).then(() => {
           this.inventory_.ack = [];
         })
       );
     }
+
     if (nacks.length) {
-      requests.push(
-        this.modifyAckDeadline_(nacks, 0).then(() => {
-          this.inventory_.nack = [];
-        })
-      );
+      const modAcks = nacks.reduce((table, [ackId, deadline]) => {
+        if (!table[deadline]) {
+          table[deadline] = [];
+        }
+
+        table[deadline].push(ackId);
+        return table;
+      }, {});
+
+      const modAckRequests = Object.keys(modAcks).map(deadline =>
+        this.modifyAckDeadline_(modAcks[deadline], Number(deadline)));
+
+      requests.push.apply(requests, modAckRequests);
+
+      Promise.all(modAckRequests).then(() => {
+        this.inventory_.nack = [];
+      });
     }
+
     return Promise.all(requests);
   }
   /*!
@@ -371,16 +388,19 @@ export class Subscriber extends EventEmitter {
    * @private
    *
    * @param {object} message - The message object.
+   * @param {number} [delay=0] - Number of seconds before the message may be redelivered
    */
-  nack_(message) {
+  nack_(message, delay = 0) {
     const breakLease = this.breakLease_.bind(this, message);
+
     if (this.isConnected_()) {
-      this.modifyAckDeadline_(message.ackId, 0, message.connectionId).then(
+      this.modifyAckDeadline_(message.ackId, delay, message.connectionId).then(
         breakLease
       );
       return;
     }
-    this.inventory_.nack.push(message.ackId);
+
+    this.inventory_.nack.push([message.ackId, delay]);
     this.setFlushTimeout_().then(breakLease);
   }
   /*!

diff --git a/test/connection-pool.ts b/test/connection-pool.ts
@@ -867,6 +867,18 @@ describe('ConnectionPool', function() {
 
       message.nack();
     });
+
+    it('should create a nack method accepting a delay argument', function(done) {
+      const delay = Math.random();
+
+      SUBSCRIPTION.nack_ = function(message_, delay_) {
+        assert.strictEqual(message_, message);
+        assert.strictEqual(delay_, delay);
+        done();
+      };
+
+      message.nack(delay);
+    });
   });
 
   describe('getAndEmitChannelState', function() {

diff --git a/test/subscriber.ts b/test/subscriber.ts
@@ -650,10 +650,12 @@ describe('Subscriber', function() {
     });
 
     it('should send any pending nacks', function() {
-      const fakeAckIds = (subscriber.inventory_.nack = ['ghi', 'jkl']);
+      const fakeAckIds = ['ghi', 'jkl'];
+
+      subscriber.inventory_.nack = fakeAckIds.map(ackId => [ackId, 0]);
 
       subscriber.modifyAckDeadline_ = function(ackIds, deadline) {
-        assert.strictEqual(ackIds, fakeAckIds);
+        assert.deepStrictEqual(ackIds, fakeAckIds);
         assert.strictEqual(deadline, 0);
         return Promise.resolve();
       };
@@ -662,6 +664,22 @@ describe('Subscriber', function() {
         assert.strictEqual(subscriber.inventory_.nack.length, 0);
       });
     });
+
+    it('should send any pending delayed nacks', function() {
+      const fakeAckIds = ['ghi', 'jkl'];
+
+      subscriber.inventory_.nack = fakeAckIds.map(ackId => [ackId, 1]);
+
+      subscriber.modifyAckDeadline_ = function(ackIds, deadline) {
+        assert.deepStrictEqual(ackIds, fakeAckIds);
+        assert.strictEqual(deadline, 1);
+        return Promise.resolve();
+      };
+
+      return subscriber.flushQueues_().then(function() {
+        assert.strictEqual(subscriber.inventory_.nack.length, 0);
+      });
+    });
   });
 
   describe('isConnected_', function() {
@@ -1045,6 +1063,18 @@ describe('Subscriber', function() {
 
         subscriber.nack_(MESSAGE);
       });
+
+      it('should use the delay if passed', function(done) {
+        subscriber.modifyAckDeadline_ = function(ackId, deadline, connId) {
+          assert.strictEqual(ackId, MESSAGE.ackId);
+          assert.strictEqual(deadline, 1);
+          assert.strictEqual(connId, MESSAGE.connectionId);
+          setImmediate(done);
+          return Promise.resolve();
+        };
+
+        subscriber.nack_(MESSAGE, 1);
+      });
     });
 
     describe('without connection', function() {
@@ -1056,7 +1086,10 @@ describe('Subscriber', function() {
 
       it('should queue the message to be nacked if no conn', function(done) {
         subscriber.setFlushTimeout_ = function() {
-          assert(subscriber.inventory_.nack.indexOf(MESSAGE.ackId) > -1);
+          assert.deepStrictEqual(
+            subscriber.inventory_.nack,
+            [[MESSAGE.ackId, 0]]
+          );
           setImmediate(done);
           return Promise.resolve();
         };
@@ -1072,6 +1105,20 @@ describe('Subscriber', function() {
 
         subscriber.nack_(MESSAGE);
       });
+
+      it('should use the delay if passed when queueing', function(done) {
+        subscriber.setFlushTimeout_ = function() {
+          assert(
+            subscriber.inventory_.nack.findIndex(element => {
+              return element[0] === MESSAGE.ackId && element[1] === 1;
+            }) > -1
+          );
+          setImmediate(done);
+          return Promise.resolve();
+        };
+
+        subscriber.nack_(MESSAGE, 1);
+      });
     });
   });