Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

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

Merged
merged 9 commits into from Jun 24, 2020
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
2b50adb
feat: Add flow control support to publisher
kamalaboulhosn Mar 24, 2020
6bb47bc
make suggested fixes
kamalaboulhosn Mar 25, 2020
0ffda7c
Merge remote-tracking branch 'upstream/master'
kamalaboulhosn May 2, 2020
9c113c3
chore: Remove note that ordering keys requires enablements.
kamalaboulhosn May 2, 2020
2a295c1
Merge remote-tracking branch 'upstream/master'
kamalaboulhosn Jun 21, 2020
6947740
feat: Add support for server-side flow control
kamalaboulhosn Jun 21, 2020
fb08aa3
Revert "chore: Remove note that ordering keys requires enablements."
kamalaboulhosn Jun 21, 2020
304059f
Merge remote-tracking branch 'upstream/master'
kamalaboulhosn Jun 24, 2020
c9c2028
fix: Fix import order
kamalaboulhosn Jun 24, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
7 changes: 7 additions & 0 deletions ...-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
View file
Open in desktop
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -93,6 +96,7 @@ public StreamingSubscriberConnection(
Distribution ackLatencyDistribution,
SubscriberStub stub,
int channelAffinity,
FlowControlSettings flowControlSettings,
FlowController flowController,
ScheduledExecutorService executor,
ScheduledExecutorService systemExecutor,
Expand All @@ -112,6 +116,7 @@ public StreamingSubscriberConnection(
executor,
systemExecutor,
clock);
this.flowControlSettings = flowControlSettings;
}

@Override
Expand Down Expand Up @@ -209,6 +214,8 @@ private void initialize() {
.setSubscription(subscription)
.setStreamAckDeadlineSeconds(60)
.setClientId(clientId)
.setMaxOutstandingMessages(flowControlSettings.getMaxOutstandingElementCount())
.setMaxOutstandingBytes(flowControlSettings.getMaxOutstandingRequestBytes())
.build());

/**
Expand Down
1 change: 1 addition & 0 deletions google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
View file
Open in desktop
Expand Up @@ -332,6 +332,7 @@ private void startStreamingConnections() {
ackLatencyDistribution,
subStub,
i,
flowControlSettings,
flowController,
executor,
alarmsExecutor,
Expand Down