Merge branch 'master' into autosynth-synthtool

feywind · web-flow · commit 0ba3b2113096 · 2020-11-18T08:17:12.000-08:00
diff --git a/src/message-stream.ts b/src/message-stream.ts
@@ -127,7 +127,6 @@ export interface MessageStreamOptions {
  * @param {MessageStreamOptions} [options] The message stream options.
  */
 export class MessageStream extends PassThrough {
-  destroyed: boolean;
   private _keepAliveHandle: NodeJS.Timer;
   private _fillHandle?: NodeJS.Timer;
   private _options: MessageStreamOptions;
@@ -139,7 +138,6 @@ export class MessageStream extends PassThrough {
 
     super({objectMode: true, highWaterMark: options.highWaterMark});
 
-    this.destroyed = false;
     this._options = options;
     this._retrier = new PullRetry();
     this._streams = new Map();
@@ -156,14 +154,24 @@ export class MessageStream extends PassThrough {
   /**
    * Destroys the stream and any underlying streams.
    *
-   * @param {error?} err An error to emit, if any.
+   * @param {error?} error An error to emit, if any.
    * @private
    */
-  destroy(err?: Error): void {
+  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.
+   *
+   * @param {error?} error An error to emit, if any.
+   * @param {Function} callback Callback for completion of any destruction.
+   * @private
+   */
+  _destroy(error: Error | null, callback: (error: Error | null) => void): void {
     this.destroyed = true;
     clearInterval(this._keepAliveHandle);
 
@@ -172,16 +180,7 @@ export class MessageStream extends PassThrough {
       stream.cancel();
     }
 
-    if (typeof super.destroy === 'function') {
-      return super.destroy(err);
-    }
-
-    process.nextTick(() => {
-      if (err) {
-        this.emit('error', err);
-      }
-      this.emit('close');
-    });
+    callback(error);
   }
   /**
    * Adds a StreamingPull stream to the combined stream.
@@ -226,8 +225,12 @@ export class MessageStream extends PassThrough {
     const request: StreamingPullRequest = {
       subscription: this._subscriber.name,
       streamAckDeadlineSeconds: this._subscriber.ackDeadline,
-      maxOutstandingMessages: this._subscriber.maxMessages,
-      maxOutstandingBytes: this._subscriber.maxBytes,
+      maxOutstandingMessages: this._subscriber.useLegacyFlowControl
+        ? 0
+        : this._subscriber.maxMessages,
+      maxOutstandingBytes: this._subscriber.useLegacyFlowControl
+        ? 0
+        : this._subscriber.maxBytes,
     };
 
     delete this._fillHandle;
diff --git a/src/subscriber.ts b/src/subscriber.ts
@@ -203,6 +203,7 @@ export interface SubscriberOptions {
   ackDeadline?: number;
   batching?: BatchOptions;
   flowControl?: FlowControlOptions;
+  useLegacyFlowControl?: boolean;
   streamingOptions?: MessageStreamOptions;
   enableOpenTelemetryTracing?: boolean;
 }
@@ -214,6 +215,9 @@ export interface SubscriberOptions {
  *     99th percentile time it takes to acknowledge a message.
  * @property {BatchOptions} [batching] Request batching options.
  * @property {FlowControlOptions} [flowControl] Flow control options.
+ * @property {boolean} [useLegacyFlowControl] Disables enforcing flow control
+ *     settings at the Cloud PubSub server and uses the less accurate method
+ *     of only enforcing flow control at the client side.
  * @property {MessageStreamOptions} [streamingOptions] Streaming options.
  */
 /**
@@ -229,6 +233,7 @@ export class Subscriber extends EventEmitter {
   ackDeadline: number;
   maxMessages: number;
   maxBytes: number;
+  useLegacyFlowControl: boolean;
   isOpen: boolean;
   private _acks!: AckQueue;
   private _histogram: Histogram;
@@ -247,6 +252,7 @@ export class Subscriber extends EventEmitter {
     this.ackDeadline = 10;
     this.maxMessages = defaultOptions.subscription.maxOutstandingMessages;
     this.maxBytes = defaultOptions.subscription.maxOutstandingBytes;
+    this.useLegacyFlowControl = false;
     this.isOpen = false;
     this._isUserSetDeadline = false;
     this._histogram = new Histogram({min: 10, max: 600});
@@ -402,6 +408,7 @@ export class Subscriber extends EventEmitter {
       this._isUserSetDeadline = true;
     }
 
+    this.useLegacyFlowControl = options.useLegacyFlowControl || false;
     if (options.flowControl) {
       this.maxMessages =
         options.flowControl!.maxMessages ||
diff --git a/test/message-stream.ts b/test/message-stream.ts
@@ -15,7 +15,7 @@
  */
 
 import * as assert from 'assert';
-import {describe, it, before, beforeEach, afterEach, after} from 'mocha';
+import {describe, it, before, beforeEach, afterEach} from 'mocha';
 import {grpc} from 'google-gax';
 import * as proxyquire from 'proxyquire';
 import * as sinon from 'sinon';
@@ -302,13 +302,7 @@ describe('MessageStream', () => {
 
   describe('destroy', () => {
     it('should noop if already destroyed', done => {
-      sandbox
-        .stub(FakePassThrough.prototype, 'destroy')
-        .callsFake(function (this: Duplex) {
-          if (this === messageStream) {
-            done();
-          }
-        });
+      messageStream.on('close', done);
 
       messageStream.destroy();
       messageStream.destroy();
@@ -350,36 +344,6 @@ describe('MessageStream', () => {
         assert.strictEqual(stub.callCount, 1);
       });
     });
-
-    describe('without native destroy', () => {
-      let destroy: (err?: Error) => void;
-
-      before(() => {
-        destroy = FakePassThrough.prototype.destroy;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        FakePassThrough.prototype.destroy = false as any;
-      });
-
-      after(() => {
-        FakePassThrough.prototype.destroy = destroy;
-      });
-
-      it('should emit close', done => {
-        messageStream.on('close', done);
-        messageStream.destroy();
-      });
-
-      it('should emit an error if present', done => {
-        const fakeError = new Error('err');
-
-        messageStream.on('error', err => {
-          assert.strictEqual(err, fakeError);
-          done();
-        });
-
-        messageStream.destroy(fakeError);
-      });
-    });
   });
 
   describe('pull stream lifecycle', () => {
