#19991

Committed 19 Sep 2025 08:23PM UTC coverage: 88.54% (-0.007%) from 88.547%

Build # #19991

Build Type

push

github

Committed by

ejona86

Commit Message

examples: Explain Bazel BCR releases and git_override option

Run Details

34665 of 39152 relevant lines covered (88.54%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.03

/../servlet/src/main/java/io/grpc/servlet/ServletAdapter.java

/*
 * Copyright 2018 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.servlet;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
import static java.util.logging.Level.FINE;
import static java.util.logging.Level.FINEST;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.io.BaseEncoding;
import io.grpc.Attributes;
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;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.ReadableBuffers;
import io.grpc.internal.ServerTransportListener;
import io.grpc.internal.StatsTraceContext;
import java.io.IOException;
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.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.logging.Logger;
import javax.servlet.AsyncContext;
import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * An adapter that transforms {@link HttpServletRequest} into gRPC request and lets a gRPC server
 * process it, and transforms the gRPC response into {@link HttpServletResponse}. An adapter can be
 * instantiated by {@link ServletServerBuilder#buildServletAdapter()}.
 *
 * <p>In a servlet, calling {@link #doPost(HttpServletRequest, HttpServletResponse)} inside {@link
 * javax.servlet.http.HttpServlet#doPost(HttpServletRequest, HttpServletResponse)} makes the servlet
 * backed by the gRPC server associated with the adapter. The servlet must support Asynchronous
 * Processing and must be deployed to a container that supports servlet 4.0 and enables HTTP/2.
 *
 * <p>The API is experimental. The authors would like to know more about the real usecases. Users
 * are welcome to provide feedback by commenting on
 * <a href=https://github.com/grpc/grpc-java/issues/5066>the tracking issue</a>.
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/5066")
public final class ServletAdapter {

  static final Logger logger = Logger.getLogger(ServletAdapter.class.getName());
  static final Function<HttpServletRequest, String> DEFAULT_METHOD_NAME_RESOLVER =
          req -> req.getRequestURI().substring(1); // remove the leading "/"

  private final ServerTransportListener transportListener;
  private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
  private final Function<HttpServletRequest, String> methodNameResolver;
  private final int maxInboundMessageSize;
  private final Attributes attributes;

  ServletAdapter(
      ServerTransportListener transportListener,
      List<? extends ServerStreamTracer.Factory> streamTracerFactories,
      Function<HttpServletRequest, String> methodNameResolver,
      int maxInboundMessageSize) {
    this.transportListener = transportListener;
    this.streamTracerFactories = streamTracerFactories;
    this.methodNameResolver = methodNameResolver;
    this.maxInboundMessageSize = maxInboundMessageSize;
    attributes = transportListener.transportReady(Attributes.EMPTY);
  }

  /**
   * Call this method inside {@link javax.servlet.http.HttpServlet#doGet(HttpServletRequest,
   * HttpServletResponse)} to serve gRPC GET request.
   *
   * <p>This method is currently not implemented.
   *
   * <p>Note that in rare case gRPC client sends GET requests.
   *
   * <p>Do not modify {@code req} and {@code resp} before or after calling this method. However,
   * calling {@code resp.setBufferSize()} before invocation is allowed.
   */
  public void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
    resp.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED, "GET method not supported");
  }

  /**
   * Call this method inside {@link javax.servlet.http.HttpServlet#doPost(HttpServletRequest,
   * HttpServletResponse)} to serve gRPC POST request.
   *
   * <p>Do not modify {@code req} and {@code resp} before or after calling this method. However,
   * calling {@code resp.setBufferSize()} before invocation is allowed.
   */
  public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
    checkArgument(req.isAsyncSupported(), "servlet does not support asynchronous operation");
    checkArgument(ServletAdapter.isGrpc(req), "the request is not a gRPC request");

    InternalLogId logId = InternalLogId.allocate(ServletAdapter.class, null);
    logger.log(FINE, "[{0}] RPC started", logId);

    AsyncContext asyncCtx = req.startAsync(req, resp);

    String method = methodNameResolver.apply(req);
    Metadata headers = getHeaders(req);

    if (logger.isLoggable(FINEST)) {
      logger.log(FINEST, "[{0}] method: {1}", new Object[] {logId, method});
      logger.log(FINEST, "[{0}] headers: {1}", new Object[] {logId, headers});
    }

    Long timeoutNanos = headers.get(TIMEOUT_KEY);
    asyncCtx.setTimeout(timeoutNanos != null
        ? TimeUnit.NANOSECONDS.toMillis(timeoutNanos) + ASYNC_TIMEOUT_SAFETY_MARGIN
        : 0);
    StatsTraceContext statsTraceCtx =
        StatsTraceContext.newServerContext(streamTracerFactories, method, headers);

    ServletServerStream stream = new ServletServerStream(
        asyncCtx,
        statsTraceCtx,
        maxInboundMessageSize,
        attributes.toBuilder()
            .set(
                Grpc.TRANSPORT_ATTR_REMOTE_ADDR,
                new InetSocketAddress(req.getRemoteHost(), req.getRemotePort()))
            .set(
                Grpc.TRANSPORT_ATTR_LOCAL_ADDR,
                new InetSocketAddress(req.getLocalAddr(), req.getLocalPort()))
            .build(),
        getAuthority(req),
        logId);

    transportListener.streamCreated(stream, method, headers);
    stream.transportState().runOnTransportThread(stream.transportState()::onStreamAllocated);

    asyncCtx.getRequest().getInputStream()
        .setReadListener(new GrpcReadListener(stream, asyncCtx, logId));
    asyncCtx.addListener(new GrpcAsyncListener(stream, logId));
  }

  /**
   * Deadlines are managed via Context, servlet async timeout is not supposed to happen.
   */
  @VisibleForTesting
  static final long ASYNC_TIMEOUT_SAFETY_MARGIN = 5_000;

  // This method must use Enumeration and its members, since that is the only way to read headers
  // from the servlet api.
  @SuppressWarnings("JdkObsolete")
  private static Metadata getHeaders(HttpServletRequest req) {
    Enumeration<String> headerNames = req.getHeaderNames();
    checkNotNull(
        headerNames, "Servlet container does not allow HttpServletRequest.getHeaderNames()");
    List<byte[]> byteArrays = new ArrayList<>();
    while (headerNames.hasMoreElements()) {
      String headerName = headerNames.nextElement();
      Enumeration<String> values = req.getHeaders(headerName);
      if (values == null) {
        continue;
      }
      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));
        } else {
          byteArrays.add(headerName.getBytes(StandardCharsets.US_ASCII));
          byteArrays.add(value.getBytes(StandardCharsets.US_ASCII));
        }
      }
    }
    return InternalMetadata.newMetadata(byteArrays.toArray(new byte[][]{}));
  }

  // This method must use HttpRequest#getRequestURL or HttpUtils#getRequestURL, both of which
  // can only return StringBuffer instances
  @SuppressWarnings("JdkObsolete")
  private static String getAuthority(HttpServletRequest req) {
    try {
      return new URI(req.getRequestURL().toString()).getAuthority();
    } catch (URISyntaxException e) {
      logger.log(FINE, "Error getting authority from the request URL {0}", req.getRequestURL());
      return req.getServerName() + ":" + req.getServerPort();
    }
  }

  /**
   * Call this method when the adapter is no longer needed. The gRPC server will be terminated.
   */
  public void destroy() {
    transportListener.transportTerminated();
  }

  private static final class GrpcAsyncListener implements AsyncListener {
    final InternalLogId logId;
    final ServletServerStream stream;

    GrpcAsyncListener(ServletServerStream stream, InternalLogId logId) {
      this.stream = stream;
      this.logId = logId;
    }

    @Override
    public void onComplete(AsyncEvent event) {
      stream.asyncCompleted = true;
    }

    @Override
    public void onTimeout(AsyncEvent event) {
      if (logger.isLoggable(FINE)) {
        logger.log(FINE, String.format("[{%s}] Timeout: ", logId), event.getThrowable());
      }
      // If the resp is not committed, cancel() to avoid being redirected to an error page.
      // Else, the container will send RST_STREAM in the end.
      if (!event.getAsyncContext().getResponse().isCommitted()) {
        stream.cancel(Status.DEADLINE_EXCEEDED);
      } else {
        stream.transportState().runOnTransportThread(
            () -> stream.transportState().transportReportStatus(Status.DEADLINE_EXCEEDED));
      }
    }

    @Override
    public void onError(AsyncEvent event) {
      if (logger.isLoggable(FINE)) {
        logger.log(FINE, String.format("[{%s}] Error: ", logId), event.getThrowable());
      }

      // If the resp is not committed, cancel() to avoid being redirected to an error page.
      // Else, the container will send RST_STREAM at the end.
      if (!event.getAsyncContext().getResponse().isCommitted()) {
        stream.cancel(Status.fromThrowable(event.getThrowable()));
      } else {
        stream.transportState().runOnTransportThread(
            () -> stream.transportState().transportReportStatus(
                Status.fromThrowable(event.getThrowable())));
      }
    }

    @Override
    public void onStartAsync(AsyncEvent event) {}
  }

  private static final class GrpcReadListener implements ReadListener {
    final ServletServerStream stream;
    final AsyncContext asyncCtx;
    final ServletInputStream input;
    final InternalLogId logId;

    GrpcReadListener(
        ServletServerStream stream,
        AsyncContext asyncCtx,
        InternalLogId logId) throws IOException {
      this.stream = stream;
      this.asyncCtx = asyncCtx;
      input = asyncCtx.getRequest().getInputStream();
      this.logId = logId;
    }

    final byte[] buffer = new byte[4 * 1024];

    @Override
    public void onDataAvailable() throws IOException {
      logger.log(FINEST, "[{0}] onDataAvailable: ENTRY", logId);

      while (input.isReady()) {
        int length = input.read(buffer);
        if (length == -1) {
          logger.log(FINEST, "[{0}] inbound data: read end of stream", logId);
          return;
        } else {
          if (logger.isLoggable(FINEST)) {
            logger.log(
                FINEST,
                "[{0}] inbound data: length = {1}, bytes = {2}",
                new Object[] {logId, length, ServletServerStream.toHexString(buffer, length)});
          }

          byte[] copy = Arrays.copyOf(buffer, length);
          stream.transportState().runOnTransportThread(
              () -> stream.transportState().inboundDataReceived(ReadableBuffers.wrap(copy), false));
        }
      }

      logger.log(FINEST, "[{0}] onDataAvailable: EXIT", logId);
    }

    @Override
    public void onAllDataRead() {
      logger.log(FINE, "[{0}] onAllDataRead", logId);
      stream.transportState().runOnTransportThread(() ->
          stream.transportState().inboundDataReceived(ReadableBuffers.empty(), true));
    }

    @Override
    public void onError(Throwable t) {
      if (logger.isLoggable(FINE)) {
        logger.log(FINE, String.format("[{%s}] Error: ", logId), t);
      }
      // If the resp is not committed, cancel() to avoid being redirected to an error page.
      // Else, the container will send RST_STREAM at the end.
      if (!asyncCtx.getResponse().isCommitted()) {
        stream.cancel(Status.fromThrowable(t));
      } else {
        stream.transportState().runOnTransportThread(
            () -> stream.transportState()
                .transportReportStatus(Status.fromThrowable(t)));
      }
    }
  }

  /**
   * Checks whether an incoming {@code HttpServletRequest} may come from a gRPC client.
   *
   * @return true if the request comes from a gRPC client
   */
  public static boolean isGrpc(HttpServletRequest request) {
    return request.getContentType() != null
        && request.getContentType().contains(GrpcUtil.CONTENT_TYPE_GRPC);
  }
}

1	/*
2	* Copyright 2018 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.servlet;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
22	import static java.util.logging.Level.FINE;
23	import static java.util.logging.Level.FINEST;
24
25	import com.google.common.annotations.VisibleForTesting;
26	import com.google.common.io.BaseEncoding;
27	import io.grpc.Attributes;
28	import io.grpc.ExperimentalApi;
29	import io.grpc.Grpc;
30	import io.grpc.InternalLogId;
31	import io.grpc.InternalMetadata;
32	import io.grpc.Metadata;
33	import io.grpc.ServerStreamTracer;
34	import io.grpc.Status;
35	import io.grpc.internal.GrpcUtil;
36	import io.grpc.internal.ReadableBuffers;
37	import io.grpc.internal.ServerTransportListener;
38	import io.grpc.internal.StatsTraceContext;
39	import java.io.IOException;
40	import java.net.InetSocketAddress;
41	import java.net.URI;
42	import java.net.URISyntaxException;
43	import java.nio.charset.StandardCharsets;
44	import java.util.ArrayList;
45	import java.util.Arrays;
46	import java.util.Enumeration;
47	import java.util.List;
48	import java.util.concurrent.TimeUnit;
49	import java.util.function.Function;
50	import java.util.logging.Logger;
51	import javax.servlet.AsyncContext;
52	import javax.servlet.AsyncEvent;
53	import javax.servlet.AsyncListener;
54	import javax.servlet.ReadListener;
55	import javax.servlet.ServletInputStream;
56	import javax.servlet.http.HttpServletRequest;
57	import javax.servlet.http.HttpServletResponse;
58
59	/**
60	* An adapter that transforms {@link HttpServletRequest} into gRPC request and lets a gRPC server
61	* process it, and transforms the gRPC response into {@link HttpServletResponse}. An adapter can be
62	* instantiated by {@link ServletServerBuilder#buildServletAdapter()}.
63	*
64	* <p>In a servlet, calling {@link #doPost(HttpServletRequest, HttpServletResponse)} inside {@link
65	* javax.servlet.http.HttpServlet#doPost(HttpServletRequest, HttpServletResponse)} makes the servlet
66	* backed by the gRPC server associated with the adapter. The servlet must support Asynchronous
67	* Processing and must be deployed to a container that supports servlet 4.0 and enables HTTP/2.
68	*
69	* <p>The API is experimental. The authors would like to know more about the real usecases. Users
70	* are welcome to provide feedback by commenting on
71	* <a href=https://github.com/grpc/grpc-java/issues/5066>the tracking issue</a>.
72	*/
73	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/5066")
74	public final class ServletAdapter {
75
76	static final Logger logger = Logger.getLogger(ServletAdapter.class.getName());	1✔
77	static final Function<HttpServletRequest, String> DEFAULT_METHOD_NAME_RESOLVER =	1✔
78	req -> req.getRequestURI().substring(1); // remove the leading "/"	1✔
79
80	private final ServerTransportListener transportListener;
81	private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
82	private final Function<HttpServletRequest, String> methodNameResolver;
83	private final int maxInboundMessageSize;
84	private final Attributes attributes;
85
86	ServletAdapter(
87	ServerTransportListener transportListener,
88	List<? extends ServerStreamTracer.Factory> streamTracerFactories,
89	Function<HttpServletRequest, String> methodNameResolver,
90	int maxInboundMessageSize) {	1✔
91	this.transportListener = transportListener;	1✔
92	this.streamTracerFactories = streamTracerFactories;	1✔
93	this.methodNameResolver = methodNameResolver;	1✔
94	this.maxInboundMessageSize = maxInboundMessageSize;	1✔
95	attributes = transportListener.transportReady(Attributes.EMPTY);	1✔
96	}	1✔
97
98	/**
99	* Call this method inside {@link javax.servlet.http.HttpServlet#doGet(HttpServletRequest,
100	* HttpServletResponse)} to serve gRPC GET request.
101	*
102	* <p>This method is currently not implemented.
103	*
104	* <p>Note that in rare case gRPC client sends GET requests.
105	*
106	* <p>Do not modify {@code req} and {@code resp} before or after calling this method. However,
107	* calling {@code resp.setBufferSize()} before invocation is allowed.
108	*/
109	public void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
110	resp.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED, "GET method not supported");	×
111	}	×
112
113	/**
114	* Call this method inside {@link javax.servlet.http.HttpServlet#doPost(HttpServletRequest,
115	* HttpServletResponse)} to serve gRPC POST request.
116	*
117	* <p>Do not modify {@code req} and {@code resp} before or after calling this method. However,
118	* calling {@code resp.setBufferSize()} before invocation is allowed.
119	*/
120	public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
121	checkArgument(req.isAsyncSupported(), "servlet does not support asynchronous operation");	1✔
122	checkArgument(ServletAdapter.isGrpc(req), "the request is not a gRPC request");	1✔
123
124	InternalLogId logId = InternalLogId.allocate(ServletAdapter.class, null);	1✔
125	logger.log(FINE, "[{0}] RPC started", logId);	1✔
126
127	AsyncContext asyncCtx = req.startAsync(req, resp);	1✔
128
129	String method = methodNameResolver.apply(req);	1✔
130	Metadata headers = getHeaders(req);	1✔
131
132	if (logger.isLoggable(FINEST)) {	1✔
133	logger.log(FINEST, "[{0}] method: {1}", new Object[] {logId, method});	×
134	logger.log(FINEST, "[{0}] headers: {1}", new Object[] {logId, headers});	×
135	}
136
137	Long timeoutNanos = headers.get(TIMEOUT_KEY);	1✔
138	asyncCtx.setTimeout(timeoutNanos != null	1✔
139	? TimeUnit.NANOSECONDS.toMillis(timeoutNanos) + ASYNC_TIMEOUT_SAFETY_MARGIN	1✔
140	: 0);	1✔
141	StatsTraceContext statsTraceCtx =	1✔
142	StatsTraceContext.newServerContext(streamTracerFactories, method, headers);	1✔
143
144	ServletServerStream stream = new ServletServerStream(	1✔
145	asyncCtx,
146	statsTraceCtx,
147	maxInboundMessageSize,
148	attributes.toBuilder()	1✔
149	.set(	1✔
150	Grpc.TRANSPORT_ATTR_REMOTE_ADDR,
151	new InetSocketAddress(req.getRemoteHost(), req.getRemotePort()))	1✔
152	.set(	1✔
153	Grpc.TRANSPORT_ATTR_LOCAL_ADDR,
154	new InetSocketAddress(req.getLocalAddr(), req.getLocalPort()))	1✔
155	.build(),	1✔
156	getAuthority(req),	1✔
157	logId);
158
159	transportListener.streamCreated(stream, method, headers);	1✔
160	stream.transportState().runOnTransportThread(stream.transportState()::onStreamAllocated);	1✔
161
162	asyncCtx.getRequest().getInputStream()	1✔
163	.setReadListener(new GrpcReadListener(stream, asyncCtx, logId));	1✔
164	asyncCtx.addListener(new GrpcAsyncListener(stream, logId));	1✔
165	}	1✔
166
167	/**
168	* Deadlines are managed via Context, servlet async timeout is not supposed to happen.
169	*/
170	@VisibleForTesting
171	static final long ASYNC_TIMEOUT_SAFETY_MARGIN = 5_000;
172
173	// This method must use Enumeration and its members, since that is the only way to read headers
174	// from the servlet api.
175	@SuppressWarnings("JdkObsolete")
176	private static Metadata getHeaders(HttpServletRequest req) {
177	Enumeration<String> headerNames = req.getHeaderNames();	1✔
178	checkNotNull(	1✔
179	headerNames, "Servlet container does not allow HttpServletRequest.getHeaderNames()");
180	List<byte[]> byteArrays = new ArrayList<>();	1✔
181	while (headerNames.hasMoreElements()) {	1✔
182	String headerName = headerNames.nextElement();	1✔
183	Enumeration<String> values = req.getHeaders(headerName);	1✔
184	if (values == null) {	1✔
185	continue;	×
186	}
187	while (values.hasMoreElements()) {	1✔
188	String value = values.nextElement();	1✔
189	if (headerName.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {	1✔
190	byteArrays.add(headerName.getBytes(StandardCharsets.US_ASCII));	1✔
191	byteArrays.add(BaseEncoding.base64().decode(value));	1✔
192	} else {
193	byteArrays.add(headerName.getBytes(StandardCharsets.US_ASCII));	1✔
194	byteArrays.add(value.getBytes(StandardCharsets.US_ASCII));	1✔
195	}
196	}	1✔
197	}	1✔
198	return InternalMetadata.newMetadata(byteArrays.toArray(new byte[][]{}));	1✔
199	}
200
201	// This method must use HttpRequest#getRequestURL or HttpUtils#getRequestURL, both of which
202	// can only return StringBuffer instances
203	@SuppressWarnings("JdkObsolete")
204	private static String getAuthority(HttpServletRequest req) {
205	try {
206	return new URI(req.getRequestURL().toString()).getAuthority();	1✔
207	} catch (URISyntaxException e) {	×
208	logger.log(FINE, "Error getting authority from the request URL {0}", req.getRequestURL());	×
209	return req.getServerName() + ":" + req.getServerPort();	×
210	}
211	}
212
213	/**
214	* Call this method when the adapter is no longer needed. The gRPC server will be terminated.
215	*/
216	public void destroy() {
217	transportListener.transportTerminated();	1✔
218	}	1✔
219
220	private static final class GrpcAsyncListener implements AsyncListener {
221	final InternalLogId logId;
222	final ServletServerStream stream;
223
224	GrpcAsyncListener(ServletServerStream stream, InternalLogId logId) {	1✔
225	this.stream = stream;	1✔
226	this.logId = logId;	1✔
227	}	1✔
228
229	@Override
230	public void onComplete(AsyncEvent event) {
231	stream.asyncCompleted = true;	1✔
232	}	1✔
233
234	@Override
235	public void onTimeout(AsyncEvent event) {
236	if (logger.isLoggable(FINE)) {	1✔
237	logger.log(FINE, String.format("[{%s}] Timeout: ", logId), event.getThrowable());	×
238	}
239	// If the resp is not committed, cancel() to avoid being redirected to an error page.
240	// Else, the container will send RST_STREAM in the end.
241	if (!event.getAsyncContext().getResponse().isCommitted()) {	1✔
242	stream.cancel(Status.DEADLINE_EXCEEDED);	1✔
243	} else {
244	stream.transportState().runOnTransportThread(	1✔
245	() -> stream.transportState().transportReportStatus(Status.DEADLINE_EXCEEDED));	1✔
246	}
247	}	1✔
248
249	@Override
250	public void onError(AsyncEvent event) {
251	if (logger.isLoggable(FINE)) {	×
252	logger.log(FINE, String.format("[{%s}] Error: ", logId), event.getThrowable());	×
253	}
254
255	// If the resp is not committed, cancel() to avoid being redirected to an error page.
256	// Else, the container will send RST_STREAM at the end.
257	if (!event.getAsyncContext().getResponse().isCommitted()) {	×
258	stream.cancel(Status.fromThrowable(event.getThrowable()));	×
259	} else {
260	stream.transportState().runOnTransportThread(	×
261	() -> stream.transportState().transportReportStatus(	×
262	Status.fromThrowable(event.getThrowable())));	×
263	}
264	}	×
265
266	@Override
267	public void onStartAsync(AsyncEvent event) {}	×
268	}
269
270	private static final class GrpcReadListener implements ReadListener {
271	final ServletServerStream stream;
272	final AsyncContext asyncCtx;
273	final ServletInputStream input;
274	final InternalLogId logId;
275
276	GrpcReadListener(
277	ServletServerStream stream,
278	AsyncContext asyncCtx,
279	InternalLogId logId) throws IOException {	1✔
280	this.stream = stream;	1✔
281	this.asyncCtx = asyncCtx;	1✔
282	input = asyncCtx.getRequest().getInputStream();	1✔
283	this.logId = logId;	1✔
284	}	1✔
285
286	final byte[] buffer = new byte[4 * 1024];	1✔
287
288	@Override
289	public void onDataAvailable() throws IOException {
290	logger.log(FINEST, "[{0}] onDataAvailable: ENTRY", logId);	1✔
291
292	while (input.isReady()) {	1✔
293	int length = input.read(buffer);	1✔
294	if (length == -1) {	1✔
295	logger.log(FINEST, "[{0}] inbound data: read end of stream", logId);	×
296	return;	×
297	} else {
298	if (logger.isLoggable(FINEST)) {	1✔
299	logger.log(	×
300	FINEST,
301	"[{0}] inbound data: length = {1}, bytes = {2}",
302	new Object[] {logId, length, ServletServerStream.toHexString(buffer, length)});	×
303	}
304
305	byte[] copy = Arrays.copyOf(buffer, length);	1✔
306	stream.transportState().runOnTransportThread(	1✔
307	() -> stream.transportState().inboundDataReceived(ReadableBuffers.wrap(copy), false));	1✔
308	}
309	}	1✔
310
311	logger.log(FINEST, "[{0}] onDataAvailable: EXIT", logId);	1✔
312	}	1✔
313
314	@Override
315	public void onAllDataRead() {
316	logger.log(FINE, "[{0}] onAllDataRead", logId);	1✔
317	stream.transportState().runOnTransportThread(() ->	1✔
318	stream.transportState().inboundDataReceived(ReadableBuffers.empty(), true));	1✔
319	}	1✔
320
321	@Override
322	public void onError(Throwable t) {
323	if (logger.isLoggable(FINE)) {	1✔
324	logger.log(FINE, String.format("[{%s}] Error: ", logId), t);	×
325	}
326	// If the resp is not committed, cancel() to avoid being redirected to an error page.
327	// Else, the container will send RST_STREAM at the end.
328	if (!asyncCtx.getResponse().isCommitted()) {	1✔
329	stream.cancel(Status.fromThrowable(t));	1✔
330	} else {
331	stream.transportState().runOnTransportThread(	×
332	() -> stream.transportState()	×
333	.transportReportStatus(Status.fromThrowable(t)));	×
334	}
335	}	1✔
336	}
337
338	/**
339	* Checks whether an incoming {@code HttpServletRequest} may come from a gRPC client.
340	*
341	* @return true if the request comes from a gRPC client
342	*/
343	public static boolean isGrpc(HttpServletRequest request) {
344	return request.getContentType() != null	1✔
345	&& request.getContentType().contains(GrpcUtil.CONTENT_TYPE_GRPC);	1✔
346	}
347	}

grpc / grpc-java / #19991

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous