-
Notifications
You must be signed in to change notification settings - Fork 80
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Better handling for uncommon/exception gRPC cases (#4016)
Fixes three logged errors. Two are handed by not permitting an error to be thrown into gRPC internals, but instead catch, cancel the stream, and log the error. The third case is handled by following the appropriate async write pattern required by servlets. Fixes #3965 Fixes #3052
- Loading branch information
Showing
5 changed files
with
296 additions
and
106 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
107 changes: 107 additions & 0 deletions
107
...a/grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/GrpcWebOutputStream.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| package io.grpc.servlet.jakarta.web; | ||
|
|
||
| import jakarta.servlet.ServletOutputStream; | ||
| import jakarta.servlet.WriteListener; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| /** | ||
| * Wraps the usual ServletOutputStream so as to allow downstream writers to use it according to the servlet spec, but | ||
| * still make it easy to write trailers as a payload instead of using HTTP trailers at the end of a stream. | ||
| */ | ||
| public class GrpcWebOutputStream extends ServletOutputStream implements WriteListener { | ||
| private final ServletOutputStream wrapped; | ||
|
|
||
| // Access to these are guarded by synchronized | ||
| private Runnable waiting; | ||
| private WriteListener writeListener; | ||
|
|
||
| public GrpcWebOutputStream(ServletOutputStream wrapped) { | ||
| this.wrapped = wrapped; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isReady() { | ||
| return wrapped.isReady(); | ||
| } | ||
|
|
||
| /** | ||
| * Internal helper method to correctly write the given bytes, then complete the stream. | ||
| * | ||
| * @param bytes the bytes to write once writing is possible | ||
| * @param close a Runnable to invoke once this write is complete | ||
| */ | ||
| public synchronized void writeAndCloseWhenReady(byte[] bytes, Runnable close) { | ||
| if (writeListener == null) { | ||
| throw new IllegalStateException("writeListener"); | ||
| } | ||
| if (isReady()) { | ||
| try { | ||
| write(bytes); | ||
| } catch (IOException ignored) { | ||
| // Ignore this error, we're closing anyway | ||
| } finally { | ||
| close.run(); | ||
| } | ||
| } else { | ||
| waiting = () -> { | ||
| try { | ||
| write(bytes); | ||
| } catch (IOException e) { | ||
| // ignore this, we're closing anyway | ||
| } finally { | ||
| close.run(); | ||
| } | ||
| }; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void setWriteListener(WriteListener writeListener) { | ||
| this.writeListener = writeListener; | ||
| wrapped.setWriteListener(this); | ||
| } | ||
|
|
||
| @Override | ||
| public void write(int i) throws IOException { | ||
| wrapped.write(i); | ||
| } | ||
|
|
||
| @Override | ||
| public void write(byte[] b) throws IOException { | ||
| wrapped.write(b); | ||
| } | ||
|
|
||
| @Override | ||
| public void write(byte[] b, int off, int len) throws IOException { | ||
| wrapped.write(b, off, len); | ||
| } | ||
|
|
||
| @Override | ||
| public void flush() throws IOException { | ||
| wrapped.flush(); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| wrapped.close(); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void onWritePossible() throws IOException { | ||
| if (writeListener != null) { | ||
| writeListener.onWritePossible(); | ||
| } | ||
| if (waiting != null) { | ||
| waiting.run(); | ||
| waiting = null; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable t) { | ||
| if (writeListener != null) { | ||
| writeListener.onError(t); | ||
| } | ||
| } | ||
| } |
70 changes: 70 additions & 0 deletions
70
...grpc-servlet-jakarta/src/main/java/io/grpc/servlet/jakarta/web/GrpcWebServletRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package io.grpc.servlet.jakarta.web; | ||
|
|
||
| import io.grpc.internal.GrpcUtil; | ||
| import jakarta.servlet.AsyncContext; | ||
| import jakarta.servlet.ServletRequest; | ||
| import jakarta.servlet.ServletResponse; | ||
| import jakarta.servlet.http.HttpServletRequest; | ||
| import jakarta.servlet.http.HttpServletRequestWrapper; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Map; | ||
| import java.util.logging.Level; | ||
| import java.util.logging.Logger; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| /** | ||
| * Wraps an incoming gRPC-web request so that a downstream filter/servlet can read it instead as a gRPC payload. This | ||
| * currently involves changing the incoming content-type, and managing the wrapped request so that downstream operations | ||
| * to handle this request behave correctly. | ||
| */ | ||
| public class GrpcWebServletRequest extends HttpServletRequestWrapper { | ||
| private static final Logger logger = Logger.getLogger(GrpcWebServletRequest.class.getName()); | ||
|
|
||
| private final GrpcWebServletResponse wrappedResponse; | ||
|
|
||
| public GrpcWebServletRequest(HttpServletRequest request, GrpcWebServletResponse wrappedResponse) { | ||
| super(request); | ||
| this.wrappedResponse = wrappedResponse; | ||
| } | ||
|
|
||
| @Override | ||
| public String getContentType() { | ||
| // Adapt the content-type to replace grpc-web with grpc | ||
| return super.getContentType().replaceFirst(Pattern.quote(GrpcWebFilter.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) { | ||
| private void safelyComplete() { | ||
| try { | ||
| // Let the superclass complete the stream so we formally close it | ||
| super.complete(); | ||
| } catch (Exception e) { | ||
| // As above, complete() should not throw, so just log this failure and continue. | ||
| // This statement is somewhat dubious, since Jetty itself is clearly throwing in | ||
| // complete()... leading us to add this try/catch to begin with. | ||
| logger.log(Level.FINE, "Error invoking complete() on underlying stream", e); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void complete() { | ||
| // Emit trailers as part of the response body, then complete the request. Note that this may mean | ||
| // that we don't actually call super.complete() synchronously. | ||
| wrappedResponse.writeTrailers(this::safelyComplete); | ||
| } | ||
| }; | ||
| } | ||
| } |
Oops, something went wrong.