fix: update to work with latest node types (changed the return type o…

…f Stream.destroy()) (#1464)

* test(nodejs): remove 15 add 16 (#1322)

Source-Link: googleapis/synthtool@6981da4
Post-Processor: gcr.io/cloud-devrel-public-resources/owlbot-nodejs:latest@sha256:3563b6b264989c4f5aa31a3682e4df36c95756cfef275d3201508947cbfc511e

* fix: update to work with latest node types (changed the return type of Stream.destroy())

* 🦉 Updates from OwlBot

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

.github/.OwlBot.lock.yaml

@@ -1,3 +1,3 @@
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-nodejs:latest
-  digest: sha256:f092066de33d4a2a13ab13c8fa9dcb4f6b96fa1fb7d391bf19cd0c4921d997c0
+  digest: sha256:3563b6b264989c4f5aa31a3682e4df36c95756cfef275d3201508947cbfc511e

.github/workflows/ci.yaml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        node: [10, 12, 14, 15]
+        node: [10, 12, 14, 16]
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-node@v1

package.json

@@ -47,7 +47,7 @@
     "precompile": "gts clean"
   },
   "dependencies": {
-    "@google-cloud/paginator": "^3.0.0",
+    "@google-cloud/paginator": "^3.0.6",
     "@google-cloud/precise-date": "^2.0.0",
     "@google-cloud/projectify": "^2.0.0",
     "@google-cloud/promisify": "^2.0.0",
@@ -72,7 +72,7 @@
     "@types/mocha": "^8.0.0",
     "@types/mv": "^2.1.0",
     "@types/ncp": "^2.0.1",
-    "@types/node": "^14.0.0",
+    "@types/node": "^16.0.0",
     "@types/proxyquire": "^1.3.28",
     "@types/sinon": "^10.0.0",
     "@types/tmp": "^0.2.0",

src/message-stream.ts

@@ -151,19 +151,6 @@ export class MessageStream extends PassThrough {
     );
     this._keepAliveHandle.unref();
   }
-  /**
-   * Destroys the stream and any underlying streams.
-   *
-   * @param {error?} error An error to emit, if any.
-   * @private
-   */
-  destroy(error?: Error | null): void {
-    // We can't assume Node has taken care of this in <14.
-    if (this.destroyed) {
-      return;
-    }
-    super.destroy(error ? error : undefined);
-  }
   /**
    * Destroys the stream and any underlying streams.
    *
@@ -172,7 +159,6 @@ export class MessageStream extends PassThrough {
    * @private
    */
   _destroy(error: Error | null, callback: (error: Error | null) => void): void {
-    this.destroyed = true;
     clearInterval(this._keepAliveHandle);
 
     for (const stream of this._streams.keys()) {
@@ -214,7 +200,8 @@ export class MessageStream extends PassThrough {
     try {
       client = await this._getClient();
     } catch (e) {
-      this.destroy(e);
+      const err = e as Error;
+      this.destroy(err);
     }
 
     if (this.destroyed) {
@@ -244,7 +231,8 @@ export class MessageStream extends PassThrough {
     try {
       await this._waitForClientReady(client);
     } catch (e) {
-      this.destroy(e);
+      const err = e as Error;
+      this.destroy(err);
     }
   }
   /**
@@ -386,7 +374,8 @@ export class MessageStream extends PassThrough {
     try {
       await promisify(client.waitForReady).call(client, deadline);
     } catch (e) {
-      throw new ChannelError(e);
+      const err = e as Error;
+      throw new ChannelError(err);
     }
   }
 }

test/message-stream.ts

@@ -38,17 +38,6 @@ const FAKE_CLIENT_CONFIG = {
   },
 };
 
-// just need this for unit tests.. we have a ponyfill for destroy on
-// MessageStream and gax streams use Duplexify
-function destroy(stream: Duplex, err?: Error): void {
-  process.nextTick(() => {
-    if (err) {
-      stream.emit('error', err);
-    }
-    stream.emit('close');
-  });
-}
-
 interface StreamState {
   highWaterMark: number;
 }
@@ -68,12 +57,6 @@ class FakePassThrough extends PassThrough {
     super(options);
     this.options = options;
   }
-  destroy(err?: Error): void {
-    if (typeof super.destroy === 'function') {
-      return super.destroy(err);
-    }
-    destroy(this, err);
-  }
 }
 
 class FakeGrpcStream extends Duplex {
@@ -95,12 +78,6 @@ class FakeGrpcStream extends Duplex {
       this.end();
     });
   }
-  destroy(err?: Error): void {
-    if (typeof super.destroy === 'function') {
-      return super.destroy(err);
-    }
-    destroy(this, err);
-  }
   _write(chunk: object, encoding: string, callback: Function): void {
     callback();
   }

test/snapshot.ts

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 import * as assert from 'assert';
+import {ServiceError} from 'google-gax';
 import {describe, it, beforeEach, before, after, afterEach} from 'mocha';
 import * as proxyquire from 'proxyquire';
 import * as sinon from 'sinon';
@@ -158,7 +159,7 @@ describe('Snapshot', () => {
           });
 
           const callback = stub.lastCall.args[2];
-          setImmediate(callback, fakeError, null, fakeResponse);
+          setImmediate(callback, fakeError as ServiceError, null, fakeResponse);
         });
 
         it('should return the correct snapshot', done => {

test/subscriber.ts

@@ -556,13 +556,16 @@ describe('Subscriber', () => {
       stream.emit('error', fakeError);
     });
 
-    it('should close the subscriber if stream closes unexpectedly', () => {
+    it('should close the subscriber if stream closes unexpectedly', done => {
       const stub = sandbox.stub(subscriber, 'close');
       const stream: FakeMessageStream = stubs.get('messageStream');
 
       stream.emit('close');
 
-      assert.strictEqual(stub.callCount, 1);
+      process.nextTick(() => {
+        assert.strictEqual(stub.callCount, 1);
+        done();
+      });
     });
 
     it('should add messages to the inventory', done => {

test/subscription.ts

@@ -393,7 +393,7 @@ describe('Subscription', () => {
       });
 
       const callback = stub.lastCall.args[3];
-      setImmediate(callback, fakeErr, null, fakeResponse);
+      setImmediate(callback, fakeErr as ServiceError, null, fakeResponse);
     });
 
     it('should update the subscription', done => {