feat: Enable server side flow control by default with the option to turn it off (#1147)

fayssalmartanigcp · web-flow · commit a9c7e0b216a0 · 2020-11-18T16:16:05.000Z
This change enables sending flow control settings automatically to the server.
If flowControl.maxMessages &gt; 0 or flowControl.maxBytes &gt; 0, flow control will
be enforced at the server side (in addition to the client side).

This behavior is enabled by default and SubscriberOptions.useLegacyFlowControl can be used for users who would
like to opt-out of this feature in case they encounter issues with server side
flow control.
diff --git a/src/message-stream.ts b/src/message-stream.ts
@@ -225,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 ||
