fix: if a flow control setting is not provided use zero (#292)

* fix: use default zero value if a flow control setting is not provided * fix lint issues
googleapis · Jul 24, 2020 · d56155b · d56155b
1 parent f54ed02
commit d56155b
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/...-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/...-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -216,8 +216,10 @@ private void initialize() {
             .setSubscription(subscription)
             .setStreamAckDeadlineSeconds(60)
             .setClientId(clientId)
-            .setMaxOutstandingMessages(flowControlSettings.getMaxOutstandingElementCount())
-            .setMaxOutstandingBytes(flowControlSettings.getMaxOutstandingRequestBytes())
+            .setMaxOutstandingMessages(
+                valueOrZero(flowControlSettings.getMaxOutstandingElementCount()))
+            .setMaxOutstandingBytes(
+                valueOrZero(flowControlSettings.getMaxOutstandingRequestBytes()))
             .build());
 
     /**
@@ -281,6 +283,10 @@ public void run() {
         MoreExecutors.directExecutor());
   }
 
+  private Long valueOrZero(Long value) {
+    return value != null ? value : 0;
+  }
+
   private boolean isAlive() {
     State state = state(); // Read the state only once.
     return state == State.RUNNING || state == State.STARTING;

diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java
@@ -19,6 +19,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.google.api.gax.batching.FlowControlSettings;
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.core.InstantiatingExecutorProvider;
@@ -238,6 +239,8 @@ private Builder getTestSubscriberBuilder(MessageReceiver receiver) {
         .setCredentialsProvider(NoCredentialsProvider.create())
         .setClock(fakeExecutor.getClock())
         .setParallelPullCount(1)
-        .setMaxDurationPerAckExtension(Duration.ofSeconds(5));
+        .setMaxDurationPerAckExtension(Duration.ofSeconds(5))
+        .setFlowControlSettings(
+            FlowControlSettings.newBuilder().setMaxOutstandingElementCount(1000L).build());
   }
 }