Skip to content

Improve grpc framing protocol #297

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

Merged
vietj merged 2 commits into from
Aug 15, 2026
Merged

Improve grpc framing protocol #297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1ef3e62
Remove trailers handling in GrpcWriteStreamBase.
vietj Aug 15, 2026
41593cf
Rework GrpcWriteStreamBase and subclasses to emit use a single action.
vietj Aug 15, 2026
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
33 changes: 25 additions & 8 deletions vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientRequestImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,8 @@ public class GrpcClientRequestImpl<Req, Resp> extends GrpcWriteStreamBase<GrpcCl
private Timer deadline;
private GrpcClientResponseImpl<Req, Resp> response;
private Handler<Void> drainHandler;
private boolean ended;
private boolean headWritten;

public GrpcClientRequestImpl(ContextInternal context,
GrpcClientInvoker invoker,
Expand Down Expand Up @@ -102,7 +104,7 @@ public GrpcClientRequest<Req, Resp> serviceName(ServiceName serviceName) {

@Override
public GrpcClientRequest<Req, Resp> fullMethodName(String fullMethodName) {
if (isHeadersSent()) {
if (isHeadersWritten()) {
throw new IllegalStateException("Request already sent");
}
int idx = fullMethodName.lastIndexOf('/');
Expand All @@ -125,7 +127,7 @@ public GrpcClientRequest<Req, Resp> timeout(long timeout, TimeUnit unit) {
if (timeout < 0L) {
throw new IllegalArgumentException("Timeout must be positive");
}
if (isHeadersSent()) {
if (isHeadersWritten()) {
throw new IllegalStateException("Timeout must be set before sending request headers");
}
String headerValue = toTimeoutHeader(timeout, unit);
Expand All @@ -149,8 +151,12 @@ public GrpcClientRequest<Req, Resp> idleTimeout(long timeout) {
}

@Override
protected Future<Void> sendHeaders(WireFormat format, String encoding, MultiMap headers) {
return sendHeaders(format, encoding, headers, false);
protected Future<Void> sendHead() {
if (headWritten) {
throw new IllegalStateException();
}
headWritten = true;
return sendHeaders(format(), encoding(), headers(), false);
}

private Future<Void> sendHeaders(WireFormat format, String encoding, MultiMap headers, boolean end) {
Expand Down Expand Up @@ -189,21 +195,32 @@ private Future<Void> sendHeaders(WireFormat format, String encoding, MultiMap he
}

@Override
protected Future<Void> sendTrailers(MultiMap trailers) {
protected Future<Void> sendEnd() {
if (stream == null) {
WireFormat wireFormat = format;
WireFormat wireFormat = format();
if (wireFormat == null) {
wireFormat = WireFormat.PROTOBUF;
format = WireFormat.PROTOBUF;
format(WireFormat.PROTOBUF);
}
return sendHeaders(wireFormat, encoding, trailers, true);
return sendHeaders(wireFormat, encoding(), null, true);
} else {
return stream.end();
}
}

@Override
protected Future<Void> sendEnd(GrpcMessage message) {
if (!headWritten) {
sendHead();
}
return stream.end(new DefaultGrpcMessageFrame(message));
}

@Override
protected Future<Void> sendMessage(GrpcMessage message) {
if (!headWritten) {
sendHead();
}
return stream.write(new DefaultGrpcMessageFrame(message));
}

Expand Down
2 changes: 1 addition & 1 deletion vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientResponseImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ public void handleEnd() {
request.cancelTimeout();
super.handleEnd();
request.handleStatus(status);
if (!request.isTrailersSent()) {
if (!request.isEndWritten()) {
request.cancel();
}
}
Expand Down
97 changes: 45 additions & 52 deletions vertx-grpc-common/src/main/java/io/vertx/grpc/common/impl/GrpcWriteStreamBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,13 @@ public abstract class GrpcWriteStreamBase<S extends GrpcWriteStreamBase<S, T>, T
protected final ContextInternal context;
private final GrpcMessageEncoder<T> messageEncoder;

protected String encoding;
protected WireFormat format;
private boolean headersSent;
private boolean trailersSent;
private String encoding;
private WireFormat format;
private boolean headersWritten;
private boolean endWritten;
private GrpcError error;
private boolean cancelled;
private MultiMap headers;
private MultiMap trailers;
private Handler<Throwable> exceptionHandler;

public GrpcWriteStreamBase(ContextInternal context, GrpcMessageEncoder<T> messageEncoder) {
Expand Down Expand Up @@ -68,37 +67,45 @@ public void cancel() {

@Override
public final S encoding(String encoding) {
if (headersSent) {
if (headersWritten) {
throw new IllegalStateException("Cannot set encoding when headers have been sent");
}
this.encoding = Objects.requireNonNull(encoding);
return (S) this;
}

public final String encoding() {
return encoding;
}

@Override
public final S format(WireFormat format) {
if (headersSent) {
if (headersWritten) {
throw new IllegalStateException("Cannot set format when headers have been sent");
}
this.format = Objects.requireNonNull(format);
return (S) this;
}

public final WireFormat format() {
return format;
}

public final ContextInternal context() {
return context;
}

public boolean isHeadersSent() {
return headersSent;
public boolean isHeadersWritten() {
return headersWritten;
}

public boolean isTrailersSent() {
return trailersSent;
public boolean isEndWritten() {
return endWritten;
}

@Override
public final MultiMap headers() {
if (headersSent) {
if (headersWritten) {
throw new IllegalStateException("Headers already sent");
}
if (headers == null) {
Expand All @@ -107,16 +114,6 @@ public final MultiMap headers() {
return headers;
}

public final MultiMap trailers() {
if (trailersSent) {
throw new IllegalStateException("Trailers already sent");
}
if (trailers == null) {
trailers = MultiMap.caseInsensitiveMultiMap();
}
return trailers;
}

@Override
public final S exceptionHandler(Handler<Throwable> handler) {
exceptionHandler = handler;
Expand Down Expand Up @@ -155,39 +152,29 @@ public final Future<Void> end() {
return writeMessage(null, true);
}

protected abstract Future<Void> sendTrailers(MultiMap trailers);
protected abstract Future<Void> sendHeaders(WireFormat wireFormat, String encoding, MultiMap headers);
protected abstract Future<Void> sendHead();
protected abstract Future<Void> sendMessage(GrpcMessage message);
protected abstract Future<Void> sendEnd(GrpcMessage message);
protected abstract Future<Void> sendEnd();
protected abstract boolean sendCancel();

private Future<Void> sendHeaders(boolean writeHeaders) {
private Future<Void> sendHead(boolean writeHeaders) {
if (!writeHeaders) {
throw new IllegalArgumentException();
}
return sendHeaders(format, encoding, headers);
}

private Future<Void> sendMessage(boolean writeHeaders, GrpcMessage message) {
if (writeHeaders) {
sendHeaders(format, encoding, headers);
}
return sendMessage(message);
}

private Future<Void> sendEnd() {
return sendTrailers(trailers);
return sendHead();
}

public final Future<Void> writeHead() {
return writeMessage(null, false);
}

protected Future<Void> writeMessage(GrpcMessage message, boolean end) {
private Future<Void> writeMessage(GrpcMessage message, boolean end) {
if (error != null) {
throw new IllegalStateException("The stream is failed: " + error);
}
if (trailersSent) {
throw new IllegalStateException("The stream has been closed");
if (end && endWritten) {
throw new IllegalStateException("The stream is ended");
}
if (message != null) {
if (format == null) {
Expand Down Expand Up @@ -231,8 +218,7 @@ protected Future<Void> writeMessage(GrpcMessage message, boolean end) {
}

boolean writeHeaders;
if (!headersSent) {
headersSent = true;
if (!headersWritten) {
writeHeaders = true;
} else {
writeHeaders = false;
Expand All @@ -241,17 +227,24 @@ protected Future<Void> writeMessage(GrpcMessage message, boolean end) {
throw new IllegalStateException();
}
}
if (end) {
trailersSent = true;
if (payload != null) {
sendMessage(writeHeaders, payload);
}
return sendEnd();
} else {
if (payload != null) {
return sendMessage(writeHeaders, payload);
try {
if (end) {
endWritten = true;
if (payload != null) {
return sendEnd(payload);
} else {
return sendEnd();
}
} else {
return sendHeaders(writeHeaders);
if (payload != null) {
return sendMessage(payload);
} else {
return sendHead(writeHeaders);
}
}
} finally {
if (writeHeaders) {
headersWritten = true;
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ public HttpConnection connection() {
}

private void handleInvocationFailure(Exception e) {
if (grpcResponse.isCancelled() || grpcResponse.isTrailersSent()) {
if (grpcResponse.isCancelled() || grpcResponse.isEndWritten()) {
context.reportException(e);
} else {
grpcResponse.fail(e);
Expand Down
43 changes: 36 additions & 7 deletions vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcServerResponseImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
import io.vertx.grpc.common.GrpcMessage;
import io.vertx.grpc.common.GrpcMessageEncoder;
import io.vertx.grpc.common.GrpcStatus;
import io.vertx.grpc.common.WireFormat;
import io.vertx.grpc.common.impl.DefaultGrpcHeadersFrame;
import io.vertx.grpc.common.impl.DefaultGrpcMessageFrame;
import io.vertx.grpc.common.impl.DefaultGrpcTrailersFrame;
Expand All @@ -44,6 +43,8 @@ public final class GrpcServerResponseImpl<Req, Resp> extends GrpcWriteStreamBase
private GrpcStatus status = GrpcStatus.OK;
private String statusMessage;
private Set<String> acceptedEncodings;
private MultiMap trailers;
private boolean headWritten;

public GrpcServerResponseImpl(ContextInternal context,
GrpcServerRequestImpl<Req, Resp> request,
Expand Down Expand Up @@ -73,7 +74,7 @@ public boolean writeQueueFull() {
}

public GrpcServerResponse<Req, Resp> status(GrpcStatus status) {
if (isTrailersSent()) {
if (isEndWritten()) {
throw new IllegalStateException("Trailers have already been sent");
}
this.status = Objects.requireNonNull(status);
Expand All @@ -82,7 +83,7 @@ public GrpcServerResponse<Req, Resp> status(GrpcStatus status) {

@Override
public GrpcServerResponse<Req, Resp> statusMessage(String msg) {
if (isTrailersSent()) {
if (isEndWritten()) {
throw new IllegalStateException("Trailers have already been sent");
}
this.statusMessage = msg;
Expand All @@ -91,7 +92,7 @@ public GrpcServerResponse<Req, Resp> statusMessage(String msg) {

public void handleTimeout() {
if (!isCancelled()) {
if (!isTrailersSent()) {
if (!isEndWritten()) {
status(GrpcStatus.DEADLINE_EXCEEDED);
end();
} else {
Expand All @@ -115,6 +116,16 @@ public GrpcStatus status() {
return status;
}

public MultiMap trailers() {
if (isEndWritten()) {
throw new IllegalStateException("Trailers already sent");
}
if (trailers == null) {
trailers = MultiMap.caseInsensitiveMultiMap();
}
return trailers;
}

@Override
public Set<String> acceptedEncodings() {
if (acceptedEncodings == null) {
Expand All @@ -137,7 +148,7 @@ public Set<String> acceptedEncodings() {
}

protected boolean sendCancel() {
if (!isTrailersSent()) {
if (!isEndWritten()) {
status(GrpcStatus.CANCELLED);
end();
return true;
Expand All @@ -153,14 +164,32 @@ protected Future<Void> sendTrailers(MultiMap grpcTrailers) {
return outbound.write(new DefaultGrpcTrailersFrame(status, statusMessage, grpcTrailers));
}

@Override
protected Future<Void> sendEnd(GrpcMessage message) {
sendMessage(message);
return sendEnd();
}

@Override
protected Future<Void> sendMessage(GrpcMessage message) {
if (!headWritten) {
sendHead();
}
return outbound.write(new DefaultGrpcMessageFrame(message));
}

@Override
protected Future<Void> sendHeaders(WireFormat wireFormat, String encoding, MultiMap headers) {
return outbound.write(new DefaultGrpcHeadersFrame(format, encoding, headers));
protected Future<Void> sendEnd() {
return sendTrailers(trailers);
}

@Override
protected Future<Void> sendHead() {
if (headWritten) {
throw new IllegalStateException();
}
headWritten = true;
return outbound.write(new DefaultGrpcHeadersFrame(format(), encoding(), headers()));
}

private static GrpcStatus mapStatus(Throwable t) {
Expand Down
Loading