feat(subscription): auto close sub on non-recoverable errors

src/subscriber.ts

@@ -251,6 +251,8 @@ export class Subscriber extends EventEmitter {
     this._inventory.clear();
 
     await this._waitForFlush();
+
+    this.emit('close');
   }
   /**
    * Gets the subscriber client instance.
@@ -309,7 +311,8 @@ export class Subscriber extends EventEmitter {
     this._stream = new MessageStream(this, streamingOptions);
 
     this._stream.on('error', err => this.emit('error', err))
-        .on('data', (data: PullResponse) => this._onData(data));
+        .on('data', (data: PullResponse) => this._onData(data))
+        .once('close', () => this.close());
 
     this._inventory.on('full', () => this._stream.pause())
         .on('free', () => this._stream.resume());

src/subscription.ts

@@ -168,6 +168,9 @@ export interface SubscriptionMetadata extends TSubscriptionMetadata {
  * // Register an error handler.
  * subscription.on('error', (err) => {});
  *
+ * // Register a close handler in case the subscriber closes unexpectedly
+ * subscription.on('close', () => {});
+ *
  * // Register a listener for `message` events.
  * function onMessage(message) {
  *   // Called every time a message is received.
@@ -251,7 +254,8 @@ export class Subscription extends EventEmitter {
 
     this._subscriber = new Subscriber(this, options as SubscriberOptions);
     this._subscriber.on('error', err => this.emit('error', err))
-        .on('message', message => this.emit('message', message));
+        .on('message', message => this.emit('message', message))
+        .on('close', () => this.emit('close'));
 
     this._listen();
   }

test/subscriber.ts

@@ -315,6 +315,11 @@ describe('Subscriber', () => {
       assert.strictEqual(stub.callCount, 1);
     });
 
+    it('should emit a close event', done => {
+      subscriber.on('close', done);
+      subscriber.close();
+    });
+
     describe('flushing the queues', () => {
       it('should wait for any pending acks', async () => {
         const ackQueue: FakeAckQueue = stubs.get('ackQueue');
@@ -476,6 +481,15 @@ describe('Subscriber', () => {
       stream.emit('error', fakeError);
     });
 
+    it('should close the subscriber if stream closes unexpectedly', () => {
+      const stub = sandbox.stub(subscriber, 'close');
+      const stream: FakeMessageStream = stubs.get('messageStream');
+
+      stream.emit('close');
+
+      assert.strictEqual(stub.callCount, 1);
+    });
+
     it('should add messages to the inventory', done => {
       subscriber.open();
 

test/subscription.ts

@@ -211,6 +211,11 @@ describe('Subscription', () => {
 
       subscriber.emit('error', error);
     });
+
+    it('should emit close events', done => {
+      subscription.on('close', done);
+      subscriber.emit('close');
+    });
   });
 
   describe('formatMetadata_', () => {