Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
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 @@ -159,6 +159,11 @@ public final void setMessageCompression(boolean enable) {
// Ignore.
}

@Override
public void setMessageCompression(boolean enabled, String compressorName) {
// Ignore.
}

@Override
public void setAuthority(String authority) {
// Ignore.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,11 @@ public final void setMessageCompression(boolean enable) {
// Ignore.
}

@Override
public void setMessageCompression(boolean enabled, String compressorName) {
// Ignore.
}

@Override
public void setAuthority(String authority) {
// Ignore.
Expand Down
23 changes: 23 additions & 0 deletions core/src/main/java/io/grpc/internal/AbstractClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,17 @@ public final void setDecompressorRegistry(DecompressorRegistry decompressorRegis
transportState().setDecompressorRegistry(decompressorRegistry);
}

/**
* {@inheritDoc}
*
* <p>Delegates to the transport state so that subclasses (e.g., NettyClientStream,
* InProcessClientStream) inherit the new implementation via their TransportState.
*/
@Override
public void setMessageCompression(boolean enabled, String compressorName) {
transportState().setMessageCompression(enabled, compressorName);
}

/** {@inheritDoc} */
@Override
protected abstract TransportState transportState();
Expand Down Expand Up @@ -261,6 +272,18 @@ private void setDecompressorRegistry(DecompressorRegistry decompressorRegistry)
checkNotNull(decompressorRegistry, "decompressorRegistry");
}

/**
* Sets whether the client is sending a gzip-compressed request. This is called by
* {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip.
* This information is used to validate the server's {@code grpc-accept-encoding} response header.
*
* @param enabled whether message compression is enabled
* @param compressorName the name of the compressor being used (e.g., "gzip")
*/
public void setMessageCompression(boolean enabled, String compressorName) {
// Default implementation does nothing. Override in Http2ClientStreamTransportState.
}

@VisibleForTesting
public final void setListener(ClientStreamListener listener) {
checkState(this.listener == null, "Already called setListener");
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/java/io/grpc/internal/ClientCallImpl.java
Original file line number Diff line number Diff line change
Expand Up @@ -542,7 +542,8 @@ private void sendMessageInternal(ReqT message) {
@Override
public void setMessageCompression(boolean enabled) {
checkState(stream != null, "Not started");
stream.setMessageCompression(enabled);
String compressorName = callOptions.getCompressor();
stream.setMessageCompression(enabled, compressorName);
}

@Override
Expand Down
10 changes: 10 additions & 0 deletions core/src/main/java/io/grpc/internal/ClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,16 @@ public interface ClientStream extends Stream {
*/
void setMaxOutboundMessageSize(int maxSize);

/**
* Sets whether the client is sending a gzip-compressed request. This is called by
* {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip.
* This information is used to validate the server's {@code grpc-accept-encoding} response header.
*
* @param enabled whether message compression is enabled
* @param compressorName the name of the compressor being used (e.g., "gzip")
*/
void setMessageCompression(boolean enabled, String compressorName);

/**
* Sets the effective deadline of the RPC.
*/
Expand Down
15 changes: 15 additions & 0 deletions core/src/main/java/io/grpc/internal/DelayedStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -447,6 +447,21 @@ public void run() {
}
}

@Override
public void setMessageCompression(boolean enabled, String compressorName) {
checkState(listener != null, "May only be called after start");
if (passThrough) {
realStream.setMessageCompression(enabled, compressorName);
} else {
delayOrExecute(new Runnable() {
@Override
public void run() {
realStream.setMessageCompression(enabled, compressorName);
}
});
}
}

@VisibleForTesting
ClientStream getRealStream() {
return realStream;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,11 @@ public void setMessageCompression(boolean enable) {
delegate().setMessageCompression(enable);
}

@Override
public void setMessageCompression(boolean enabled, String compressorName) {
delegate().setMessageCompression(enabled, compressorName);
}

@Override
public void cancel(Status reason) {
delegate().cancel(reason);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@

package io.grpc.internal;

import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER;
import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;

import com.google.common.base.Preconditions;
import io.grpc.CallOptions;
import io.grpc.InternalMetadata;
Expand All @@ -24,6 +27,8 @@
import io.grpc.Status;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
Expand Down Expand Up @@ -65,6 +70,14 @@ public Integer parseAsciiString(byte[] serialized) {
private Charset errorCharset = StandardCharsets.UTF_8;
private boolean headersReceived;

/**
* Tracks whether the client sent a gzip-encoded request. This is set by {@link
* #setMessageCompression(boolean, String)} when the compressor is gzip.
*/
private boolean clientSentGzipRequest = false;

private static final Logger log = Logger.getLogger(Http2ClientStreamTransportState.class.getName());

protected Http2ClientStreamTransportState(
int maxMessageSize,
StatsTraceContext statsTraceCtx,
Expand All @@ -73,6 +86,21 @@ protected Http2ClientStreamTransportState(
super(maxMessageSize, statsTraceCtx, transportTracer, options);
}

/**
* Sets whether the client is sending a gzip-compressed request. This is called by
* {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip.
* This information is used to validate the server's {@code grpc-accept-encoding} response header.
*
* @param enabled whether message compression is enabled
* @param compressorName the name of the compressor being used (e.g., "gzip")
*/
@Override
public final void setMessageCompression(boolean enabled, String compressorName) {
if (enabled && "gzip".equals(compressorName)) {
clientSentGzipRequest = true;
}
}

/**
* Called to process a failure in HTTP/2 processing. It should notify the transport to cancel the
* stream and call {@code transportReportStatus()}.
Expand Down Expand Up @@ -109,6 +137,9 @@ protected void transportHeadersReceived(Metadata headers) {
return;
}

// Validate grpc-accept-encoding header if client sent gzip request
validateGrpcAcceptEncoding(headers);

stripTransportDetails(headers);
inboundHeadersReceived(headers);
} finally {
Expand Down Expand Up @@ -257,4 +288,47 @@ private static void stripTransportDetails(Metadata metadata) {
metadata.discardAll(InternalStatus.CODE_KEY);
metadata.discardAll(InternalStatus.MESSAGE_KEY);
}
}

/**
* Validates that the server's response includes a {@code grpc-accept-encoding} header that
* includes {@code gzip} when the client sent a gzip-encoded request.
*
* <p>According to the gRPC spec, when a client sends a gzip-encoded request, the server must
* respond with {@code grpc-accept-encoding: gzip} in the response headers to indicate it can
* accept gzip-encoded responses. If this header is missing or doesn't include gzip, it's a
* server misbehavior that we log at FINE level.
*
* @param headers the response headers from the server
*/
private void validateGrpcAcceptEncoding(Metadata headers) {
if (!clientSentGzipRequest) {
// No validation needed if client didn't send gzip
return;
}

byte[] acceptEncodingBytes = headers.get(MESSAGE_ACCEPT_ENCODING_KEY);
if (acceptEncodingBytes == null) {
log.log(Level.FINE,
"Server sent gzip-encoded request but response missing grpc-accept-encoding header. "
+ "This is server misbehavior.");
return;
}

String acceptEncoding = new String(acceptEncodingBytes, StandardCharsets.US_ASCII);
// Check if gzip is in the accepted encodings (comma-separated list)
Iterable<String> encodingsIterable = GrpcUtil.ACCEPT_ENCODING_SPLITTER.split(acceptEncoding);
boolean gzipAccepted = false;
for (String encoding : encodingsIterable) {
if ("gzip".equalsIgnoreCase(encoding.trim())) {
gzipAccepted = true;
break;
}
}

if (!gzipAccepted) {
log.log(Level.FINE,
"Server sent gzip-encoded request but grpc-accept-encoding ({0}) does not include gzip. "
+ "This is server misbehavior.", acceptEncoding);
}
}
}
5 changes: 5 additions & 0 deletions core/src/main/java/io/grpc/internal/NoopClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,11 @@ public void setMessageCompression(boolean enable) {
// noop
}

@Override
public void setMessageCompression(boolean enabled, String compressorName) {
// noop
}

@Override
public void optimizeForDirectExecutor() {}

Expand Down
12 changes: 12 additions & 0 deletions core/src/main/java/io/grpc/internal/RetriableStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -689,6 +689,18 @@ public void runWith(Substream substream) {
delayOrExecute(new MessageCompressionEntry());
}

@Override
public final void setMessageCompression(boolean enabled, String compressorName) {
class MessageCompressionEntry implements BufferEntry {
@Override
public void runWith(Substream substream) {
substream.stream.setMessageCompression(enabled, compressorName);
}
}

delayOrExecute(new MessageCompressionEntry());
}

@Override
public final void halfClose() {
class HalfCloseEntry implements BufferEntry {
Expand Down
Loading
Loading