Skip to content
This repository was archived by the owner on Apr 7, 2026. It is now read-only.
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,12 @@ static final class KeyAwareClientCall<RequestT, ResponseT>
@Nullable private ChannelEndpoint selectedEndpoint;
@Nullable private ByteString transactionIdToClear;
private boolean allowDefaultAffinity;
private long pendingRequests;
private boolean pendingHalfClose;
@Nullable private Boolean pendingMessageCompression;
private boolean cancelled;
@Nullable private String cancelMessage;
@Nullable private Throwable cancelCause;
Comment thread
rahul2393 marked this conversation as resolved.
Outdated

KeyAwareClientCall(
KeyAwareChannel parentChannel,
Expand All @@ -268,11 +274,22 @@ protected ClientCall<RequestT, ResponseT> delegate() {
public void start(Listener<ResponseT> responseListener, Metadata headers) {
this.responseListener = new KeyAwareClientCallListener<>(responseListener, this);
this.headers = headers;
if (cancelled) {
this.responseListener.onClose(
io.grpc.Status.CANCELLED.withDescription(cancelMessage).withCause(cancelCause),
new Metadata());
}
}

@Override
@SuppressWarnings("unchecked")
public void sendMessage(RequestT message) {
if (cancelled) {
return;
}
if (responseListener == null || headers == null) {
throw new IllegalStateException("start must be called before sendMessage");
}
ChannelEndpoint endpoint = null;
ChannelFinder finder = null;

Expand Down Expand Up @@ -326,16 +343,23 @@ public void sendMessage(RequestT message) {
this.channelFinder = finder;

delegate = endpoint.getChannel().newCall(methodDescriptor, callOptions);
if (pendingMessageCompression != null) {
delegate.setMessageCompression(pendingMessageCompression);
}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this.pendingMessageCompression is never set back to null, meaning that once it has been set, this if statement will be true for every invocation.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated to set it to null

delegate.start(responseListener, headers);
drainPendingRequests();
delegate.sendMessage(message);
if (pendingHalfClose) {
delegate.halfClose();
}
}

@Override
public void halfClose() {
if (delegate != null) {
delegate.halfClose();
} else {
throw new IllegalStateException("halfClose called before sendMessage");
pendingHalfClose = true;
}
}

Expand All @@ -346,6 +370,56 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) {
} else if (responseListener != null) {
responseListener.onClose(
io.grpc.Status.CANCELLED.withDescription(message).withCause(cause), new Metadata());
cancelled = true;
cancelMessage = message;
cancelCause = cause;
} else {
cancelled = true;
cancelMessage = message;
cancelCause = cause;
}
Comment thread
rahul2393 marked this conversation as resolved.
Outdated
}

@Override
public void request(int numMessages) {
if (delegate != null) {
Comment thread
rahul2393 marked this conversation as resolved.
Outdated
delegate.request(numMessages);
return;
}
if (numMessages <= 0) {
return;
}
long updated = pendingRequests + numMessages;
if (updated < 0L) {
updated = Long.MAX_VALUE;
}
pendingRequests = updated;
}

@Override
public boolean isReady() {
if (delegate == null) {
return false;
}
return delegate.isReady();
}

@Override
public void setMessageCompression(boolean enabled) {
if (delegate != null) {
delegate.setMessageCompression(enabled);
} else {
pendingMessageCompression = enabled;
}
}

private void drainPendingRequests() {
long requests = pendingRequests;
pendingRequests = 0L;
while (requests > 0) {
int batch = requests > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) requests;
delegate.request(batch);
requests -= batch;
}
}

Expand Down
Loading