fix: ignore messages that come in after close (#485)

callmehiphop · JustinBeckwith · commit e59f8ecf47ef · 2019-02-17T19:29:47.000-08:00
diff --git a/src/lease-manager.ts b/src/lease-manager.ts
@@ -216,7 +216,9 @@ export class LeaseManager extends EventEmitter {
    * @param {Message} message The message to emit.
    */
   private _dispense(message: Message): void {
-    process.nextTick(() => this._subscriber.emit('message', message));
+    if (this._subscriber.isOpen) {
+      process.nextTick(() => this._subscriber.emit('message', message));
+    }
   }
   /**
    * Loops through inventory and extends the deadlines for any messages that
diff --git a/src/subscriber.ts b/src/subscriber.ts
@@ -365,13 +365,17 @@ export class Subscriber extends EventEmitter {
    *
    * @private
    */
-  private _onData(response: PullResponse): void {
-    response.receivedMessages.forEach((data: ReceivedMessage) => {
+  private _onData({receivedMessages}: PullResponse): void {
+    for (const data of receivedMessages) {
       const message = new Message(this, data);
 
-      message.modAck(this.ackDeadline);
-      this._inventory.add(message);
-    });
+      if (this.isOpen) {
+        message.modAck(this.ackDeadline);
+        this._inventory.add(message);
+      } else {
+        message.nack();
+      }
+    }
   }
 
   /**
diff --git a/system-test/pubsub.ts b/system-test/pubsub.ts
@@ -415,11 +415,8 @@ describe('pubsub', () => {
       subscription.on('message', ack);
 
       function ack(message) {
-        // remove listener to we only ack first message
-        subscription.removeListener('message', ack);
-
         message.ack();
-        setTimeout(() => subscription.close(done), 2500);
+        subscription.close(done);
       }
     });
 
@@ -430,11 +427,8 @@ describe('pubsub', () => {
       subscription.on('message', nack);
 
       function nack(message) {
-        // remove listener to we only ack first message
-        subscription.removeListener('message', nack);
-
         message.nack();
-        setTimeout(() => subscription.close(done), 2500);
+        subscription.close(done);
       }
     });
 
@@ -588,36 +582,32 @@ describe('pubsub', () => {
     let subscription;
     let snapshot;
 
-    function deleteAllSnapshots() {
-      // tslint:disable-next-line no-any
-      return (pubsub.getSnapshots() as any).then(data => {
-        return Promise.all(data[0].map(snapshot => {
-          return snapshot.delete();
-        }));
-      });
+    function getSnapshotName({name}) {
+      return name.split('/').pop();
     }
 
     before(async () => {
       topic = pubsub.topic(generateTopicName());
       subscription = topic.subscription(generateSubName());
       snapshot = subscription.snapshot(SNAPSHOT_NAME);
 
-      await deleteAllSnapshots();
       await topic.create();
       await subscription.create();
       await snapshot.create();
     });
 
     after(async () => {
-      await deleteAllSnapshots();
+      await snapshot.delete();
+      await subscription.delete();
       await topic.delete();
     });
 
     it('should get a list of snapshots', done => {
       pubsub.getSnapshots((err, snapshots) => {
         assert.ifError(err);
-        assert.strictEqual(snapshots.length, 1);
-        assert.strictEqual(snapshots[0].name.split('/').pop(), SNAPSHOT_NAME);
+        assert(snapshots.length > 0);
+        const names = snapshots.map(getSnapshotName);
+        assert(names.includes(SNAPSHOT_NAME));
         done();
       });
     });
@@ -633,9 +623,9 @@ describe('pubsub', () => {
                 snapshots.push(snapshot);
               })
           .on('end', () => {
-            assert.strictEqual(snapshots.length, 1);
-            assert.strictEqual(
-                snapshots[0].name.split('/').pop(), SNAPSHOT_NAME);
+            assert(snapshots.length > 0);
+            const names = snapshots.map(getSnapshotName);
+            assert(names.includes(SNAPSHOT_NAME));
             done();
           });
     });
diff --git a/test/lease-manager.ts b/test/lease-manager.ts
@@ -26,6 +26,7 @@ const fakeos = {
 
 class FakeSubscriber extends EventEmitter {
   ackDeadline = 10;
+  isOpen = true;
   modAckLatency = 2000;
   async modAck(message: FakeMessage, deadline: number): Promise<void> {}
 }
@@ -140,7 +141,7 @@ describe('LeaseManager', () => {
     });
 
     it('should not dispatch the message if the inventory is full', done => {
-      const message = new FakeMessage();
+      const fakeMessage = new FakeMessage();
 
       leaseManager.isFull = () => true;
       leaseManager.setOptions({allowExcessMessages: false});
@@ -149,6 +150,21 @@ describe('LeaseManager', () => {
         done(new Error('Test should not have dispatched message.'));
       });
 
+      leaseManager.add(fakeMessage);
+      setImmediate(done);
+    });
+
+    it('should not dispatch the message if the sub closes', done => {
+      const fakeMessage = new FakeMessage();
+
+      leaseManager.isFull = () => false;
+
+      subscriber.isOpen = false;
+      subscriber.on('message', () => {
+        done(new Error('Test should not have dispatched message.'));
+      });
+
+      leaseManager.add(fakeMessage);
       setImmediate(done);
     });
 
diff --git a/test/subscriber.ts b/test/subscriber.ts
@@ -320,6 +320,18 @@ describe('Subscriber', () => {
       subscriber.close();
     });
 
+    it('should nack any messages that come in after', () => {
+      const stream: FakeMessageStream = stubs.get('messageStream');
+      const stub = sandbox.stub(subscriber, 'nack');
+      const pullResponse = {receivedMessages: [RECEIVED_MESSAGE]};
+
+      subscriber.close();
+      stream.emit('data', pullResponse);
+
+      const [{ackId}] = stub.lastCall.args;
+      assert.strictEqual(ackId, RECEIVED_MESSAGE.ackId);
+    });
+
     describe('flushing the queues', () => {
       it('should wait for any pending acks', async () => {
         const ackQueue: FakeAckQueue = stubs.get('ackQueue');
