diff --git a/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/ServletAdapter.java b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/ServletAdapter.java index f52afb0fd3f..99d66a624f1 100644 --- a/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/ServletAdapter.java +++ b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/ServletAdapter.java @@ -18,7 +18,6 @@ import io.grpc.ExperimentalApi; import io.grpc.Grpc; import io.grpc.InternalLogId; -import io.grpc.InternalMetadata; import io.grpc.Metadata; import io.grpc.ServerStreamTracer; import io.grpc.Status; @@ -38,12 +37,12 @@ import java.net.InetSocketAddress; import java.net.URI; import java.net.URISyntaxException; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; import java.util.Arrays; import java.util.Enumeration; import java.util.List; +import java.util.Locale; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Logger; import static com.google.common.base.Preconditions.checkArgument; @@ -180,25 +179,26 @@ private static Metadata getHeaders(HttpServletRequest req) { Enumeration headerNames = req.getHeaderNames(); checkNotNull( headerNames, "Servlet container does not allow HttpServletRequest.getHeaderNames()"); - List byteArrays = new ArrayList<>(); + Metadata metadata = new Metadata(); while (headerNames.hasMoreElements()) { String headerName = headerNames.nextElement(); Enumeration values = req.getHeaders(headerName); if (values == null) { continue; } + headerName = headerName.toLowerCase(Locale.ROOT); + boolean isBinaryHeader = headerName.endsWith(Metadata.BINARY_HEADER_SUFFIX); while (values.hasMoreElements()) { String value = values.nextElement(); - if (headerName.endsWith(Metadata.BINARY_HEADER_SUFFIX)) { - byteArrays.add(headerName.getBytes(StandardCharsets.US_ASCII)); - byteArrays.add(BaseEncoding.base64().decode(value)); + if (isBinaryHeader) { + metadata.put(Metadata.Key.of(headerName, Metadata.BINARY_BYTE_MARSHALLER), + BaseEncoding.base64().decode(value)); } else { - byteArrays.add(headerName.getBytes(StandardCharsets.US_ASCII)); - byteArrays.add(value.getBytes(StandardCharsets.US_ASCII)); + metadata.put(Metadata.Key.of(headerName, Metadata.ASCII_STRING_MARSHALLER), value); } } } - return InternalMetadata.newMetadata(byteArrays.toArray(new byte[][] {})); + return metadata; } // This method must use HttpRequest#getRequestURL or HttpUtils#getRequestURL, both of which @@ -273,6 +273,7 @@ private static final class GrpcReadListener implements ReadListener { final AsyncContext asyncCtx; final ServletInputStream input; final InternalLogId logId; + private final AtomicBoolean closed = new AtomicBoolean(false); GrpcReadListener( ServletServerStream stream, @@ -315,6 +316,11 @@ public void onDataAvailable() throws IOException { @Override public void onAllDataRead() { logger.log(FINE, "[{0}] onAllDataRead", logId); + if (!closed.compareAndSet(false, true)) { + // https://github.com/eclipse/jetty.project/issues/8405 + logger.log(FINE, "[{0}] onAllDataRead already called, skipping this one", logId); + return; + } stream.transportState().runOnTransportThread( () -> stream.transportState().inboundDataReceived(ReadableBuffers.empty(), true)); } diff --git a/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/DelegatingAsyncContext.java b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/DelegatingAsyncContext.java new file mode 100644 index 00000000000..a51577f99a8 --- /dev/null +++ b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/DelegatingAsyncContext.java @@ -0,0 +1,88 @@ +package io.grpc.servlet.jakarta.web; + +import jakarta.servlet.AsyncContext; +import jakarta.servlet.AsyncListener; +import jakarta.servlet.ServletContext; +import jakarta.servlet.ServletException; +import jakarta.servlet.ServletRequest; +import jakarta.servlet.ServletResponse; + +/** + * Util class to allow the complete() call to get some work done (writing trailers as a payload) before calling the + * actual container implementation. The container will finish closing the stream before invoking the async listener and + * formally informing the filter that the stream has closed, making this our last chance to intercept the closing of the + * stream before it happens. + */ +public class DelegatingAsyncContext implements AsyncContext { + private final AsyncContext delegate; + + public DelegatingAsyncContext(AsyncContext delegate) { + this.delegate = delegate; + } + + @Override + public ServletRequest getRequest() { + return delegate.getRequest(); + } + + @Override + public ServletResponse getResponse() { + return delegate.getResponse(); + } + + @Override + public boolean hasOriginalRequestAndResponse() { + return delegate.hasOriginalRequestAndResponse(); + } + + @Override + public void dispatch() { + delegate.dispatch(); + } + + @Override + public void dispatch(String path) { + delegate.dispatch(path); + } + + @Override + public void dispatch(ServletContext context, String path) { + delegate.dispatch(context, path); + } + + @Override + public void complete() { + delegate.complete(); + } + + @Override + public void start(Runnable run) { + delegate.start(run); + } + + @Override + public void addListener(AsyncListener listener) { + delegate.addListener(listener); + } + + @Override + public void addListener(AsyncListener listener, ServletRequest servletRequest, + ServletResponse servletResponse) { + delegate.addListener(listener, servletRequest, servletResponse); + } + + @Override + public T createListener(Class clazz) throws ServletException { + return delegate.createListener(clazz); + } + + @Override + public void setTimeout(long timeout) { + delegate.setTimeout(timeout); + } + + @Override + public long getTimeout() { + return delegate.getTimeout(); + } +} diff --git a/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/GrpcWebFilter.java b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/GrpcWebFilter.java new file mode 100644 index 00000000000..66fba1f93a9 --- /dev/null +++ b/grpc-java/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/GrpcWebFilter.java @@ -0,0 +1,134 @@ +package io.grpc.servlet.jakarta.web; + +import io.grpc.internal.GrpcUtil; +import jakarta.servlet.AsyncContext; +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import jakarta.servlet.ServletRequest; +import jakarta.servlet.ServletResponse; +import jakarta.servlet.http.HttpFilter; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; +import jakarta.servlet.http.HttpServletResponse; +import jakarta.servlet.http.HttpServletResponseWrapper; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; +import java.util.regex.Pattern; + +/** + * Servlet filter that translates grpc-web on the fly to match what is expected by GrpcServlet. This work is done + * in-process with no addition copies to the request or response data - only the content type header and the trailer + * content is specially treated at this time. + * + * Note that grpc-web-text is not yet supported. + */ +public class GrpcWebFilter extends HttpFilter { + private static final Logger logger = Logger.getLogger(GrpcWebFilter.class.getName()); + + public static final String CONTENT_TYPE_GRPC_WEB = GrpcUtil.CONTENT_TYPE_GRPC + "-web"; + + @Override + public void doFilter(HttpServletRequest request, HttpServletResponse response, FilterChain chain) + throws IOException, ServletException { + if (isGrpcWeb(request)) { + // wrap the request and response to paper over the grpc-web details + GrpcWebHttpResponse wrappedResponse = new GrpcWebHttpResponse(response); + HttpServletRequestWrapper wrappedRequest = new HttpServletRequestWrapper(request) { + @Override + public String getContentType() { + // Adapt the content-type to replace grpc-web with grpc + return super.getContentType().replaceFirst(Pattern.quote(CONTENT_TYPE_GRPC_WEB), + GrpcUtil.CONTENT_TYPE_GRPC); + } + + @Override + public AsyncContext startAsync() throws IllegalStateException { + return startAsync(this, wrappedResponse); + } + + @Override + public AsyncContext startAsync(ServletRequest servletRequest, ServletResponse servletResponse) + throws IllegalStateException { + AsyncContext delegate = super.startAsync(servletRequest, servletResponse); + return new DelegatingAsyncContext(delegate) { + @Override + public void complete() { + // Write any trailers out to the output stream as a payload, since grpc-web doesn't + // use proper trailers. + try { + if (wrappedResponse.trailers != null) { + Map map = wrappedResponse.trailers.get(); + if (map != null) { + // write a payload, even for an empty set of trailers, but not for + // the absence of trailers. + int trailerLength = map.entrySet().stream() + .mapToInt(e -> e.getKey().length() + e.getValue().length() + 4).sum(); + ByteBuffer payload = ByteBuffer.allocate(5 + trailerLength); + payload.put((byte) 0x80); + payload.putInt(trailerLength); + for (Map.Entry entry : map.entrySet()) { + payload.put(entry.getKey().getBytes(StandardCharsets.US_ASCII)); + payload.put(": ".getBytes(StandardCharsets.US_ASCII)); + payload.put(entry.getValue().getBytes(StandardCharsets.US_ASCII)); + payload.put("\r\n".getBytes(StandardCharsets.US_ASCII)); + } + wrappedResponse.getOutputStream().write(payload.array()); + } + } + } catch (IOException e) { + // complete() should not throw, but instead just log the error. In this case, + // the connection has likely been lost, so there is no way to send the trailers, + // so we just let the exception slide. + logger.log(Level.FINE, "Error sending grpc-web trailers", e); + } + + // Let the superclass complete the stream so we formally close it + super.complete(); + } + }; + } + }; + + chain.doFilter(wrappedRequest, wrappedResponse); + } else { + chain.doFilter(request, response); + } + } + + private static boolean isGrpcWeb(ServletRequest request) { + return request.getContentType() != null && request.getContentType().startsWith(CONTENT_TYPE_GRPC_WEB); + } + + // Technically we should throw away content-length too, but the impl won't care + public static class GrpcWebHttpResponse extends HttpServletResponseWrapper { + private Supplier> trailers; + + public GrpcWebHttpResponse(HttpServletResponse response) { + super(response); + } + + @Override + public void setContentType(String type) { + // Adapt the content-type to be grpc-web + super.setContentType( + type.replaceFirst(Pattern.quote(GrpcUtil.CONTENT_TYPE_GRPC), CONTENT_TYPE_GRPC_WEB)); + } + + // intercept trailers and write them out as a message just before we complete + @Override + public void setTrailerFields(Supplier> supplier) { + trailers = supplier; + } + + @Override + public Supplier> getTrailerFields() { + return trailers; + } + } +} diff --git a/server/jetty/src/main/java/io/deephaven/server/jetty/JettyBackedGrpcServer.java b/server/jetty/src/main/java/io/deephaven/server/jetty/JettyBackedGrpcServer.java index ed72f5758c3..5e91fbe9290 100644 --- a/server/jetty/src/main/java/io/deephaven/server/jetty/JettyBackedGrpcServer.java +++ b/server/jetty/src/main/java/io/deephaven/server/jetty/JettyBackedGrpcServer.java @@ -11,6 +11,7 @@ import io.deephaven.ssl.config.TrustJdk; import io.deephaven.ssl.config.impl.KickstartUtils; import io.grpc.servlet.web.websocket.WebSocketServerStream; +import io.grpc.servlet.jakarta.web.GrpcWebFilter; import jakarta.servlet.DispatcherType; import jakarta.websocket.server.ServerEndpointConfig; import nl.altindag.ssl.SSLFactory; @@ -78,6 +79,9 @@ public JettyBackedGrpcServer( // Direct jetty all use this configuration as the root application context.setContextPath("/"); + // Handle grpc-web connections, translate to vanilla grpc + context.addFilter(new FilterHolder(new GrpcWebFilter()), "/*", EnumSet.noneOf(DispatcherType.class)); + // Wire up the provided grpc filter context.addFilter(new FilterHolder(filter), "/*", EnumSet.noneOf(DispatcherType.class));