feat: Send streaming pull flow control settings to server (googleapis#267)

kamalaboulhosn · web-flow · commit 9c750c8e4899 · 2020-06-24T13:41:56.000-04:00
* feat: Add flow control support to publisher * make suggested fixes * chore: Remove note that ordering keys requires enablements. * feat: Add support for server-side flow control * Revert "chore: Remove note that ordering keys requires enablements." This reverts commit 9c113c3. * fix: Fix import order
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -24,6 +24,7 @@
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.batching.FlowControlSettings;
 import com.google.api.gax.batching.FlowController;
 import com.google.api.gax.core.Distribution;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -71,6 +72,8 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
   private final ScheduledExecutorService systemExecutor;
   private final MessageDispatcher messageDispatcher;
 
+  private final FlowControlSettings flowControlSettings;
+
   private final AtomicLong channelReconnectBackoffMillis =
       new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
   private final Waiter ackOperationsWaiter = new Waiter();
@@ -93,6 +96,7 @@ public StreamingSubscriberConnection(
       Distribution ackLatencyDistribution,
       SubscriberStub stub,
       int channelAffinity,
+      FlowControlSettings flowControlSettings,
       FlowController flowController,
       ScheduledExecutorService executor,
       ScheduledExecutorService systemExecutor,
@@ -112,6 +116,7 @@ public StreamingSubscriberConnection(
             executor,
             systemExecutor,
             clock);
+    this.flowControlSettings = flowControlSettings;
   }
 
   @Override
@@ -209,6 +214,8 @@ private void initialize() {
             .setSubscription(subscription)
             .setStreamAckDeadlineSeconds(60)
             .setClientId(clientId)
+            .setMaxOutstandingMessages(flowControlSettings.getMaxOutstandingElementCount())
+            .setMaxOutstandingBytes(flowControlSettings.getMaxOutstandingRequestBytes())
             .build());
 
     /**
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
@@ -332,6 +332,7 @@ private void startStreamingConnections() {
                 ackLatencyDistribution,
                 subStub,
                 i,
+                flowControlSettings,
                 flowController,
                 executor,
                 alarmsExecutor,
