From 5328718fc1a670b51739117cf0a6aee4aa9ece16 Mon Sep 17 00:00:00 2001 From: gregw Date: Tue, 26 Sep 2023 11:59:09 +1000 Subject: [PATCH 1/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- invoker/core/pom.xml | 22 ++- .../invoker/BackgroundFunctionExecutor.java | 54 ++++--- .../invoker/HttpFunctionExecutor.java | 29 ++-- .../invoker/TypedFunctionExecutor.java | 32 ++-- .../invoker/http/HttpRequestImpl.java | 128 +++++++++------ .../invoker/http/HttpResponseImpl.java | 82 +++++----- .../functions/invoker/runner/Invoker.java | 71 +++++---- .../functions/invoker/IntegrationTest.java | 50 +++--- .../functions/invoker/http/HttpTest.java | 147 ++++++++++-------- 9 files changed, 357 insertions(+), 258 deletions(-) diff --git a/invoker/core/pom.xml b/invoker/core/pom.xml index 9d809a6a..baff09df 100644 --- a/invoker/core/pom.xml +++ b/invoker/core/pom.xml @@ -20,9 +20,10 @@ UTF-8 5.3.2 - 11 - 11 + 17 + 17 2.5.0 + 12.0.2-SNAPSHOT @@ -46,11 +47,6 @@ functions-framework-api 1.1.0 - - javax.servlet - javax.servlet-api - 4.0.1 - io.cloudevents cloudevents-core @@ -97,13 +93,13 @@ org.eclipse.jetty - jetty-servlet - 9.4.52.v20230823 + jetty-server + ${jetty.version} - org.eclipse.jetty - jetty-server - 9.4.52.v20230823 + org.slf4j + slf4j-jdk14 + 2.0.9 com.beust @@ -151,7 +147,7 @@ org.eclipse.jetty jetty-client - 9.4.52.v20230823 + ${jetty.version} test diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java index 98b9bc8a..cd07eaa6 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java @@ -29,25 +29,32 @@ import io.cloudevents.http.HttpMessageFactory; import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; import java.io.Reader; import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; import java.time.OffsetDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.TreeMap; import java.util.logging.Level; import java.util.logging.Logger; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpField; +import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.http.HttpStatus; +import org.eclipse.jetty.io.Content; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.util.Callback; /** Executes the user's background function. */ -public final class BackgroundFunctionExecutor extends HttpServlet { +public final class BackgroundFunctionExecutor extends Handler.Abstract { private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker"); private final FunctionExecutor functionExecutor; @@ -175,8 +182,10 @@ static Optional backgroundFunctionTypeArgument( .findFirst(); } - private static Event parseLegacyEvent(HttpServletRequest req) throws IOException { - try (BufferedReader bodyReader = req.getReader()) { + private static Event parseLegacyEvent(Request req) throws IOException { + try (BufferedReader bodyReader = new BufferedReader( + new InputStreamReader(Content.Source.asInputStream(req), + Objects.requireNonNullElse(Request.getCharset(req), StandardCharsets.ISO_8859_1)))) { return parseLegacyEvent(bodyReader); } } @@ -223,7 +232,7 @@ private static Context contextFromCloudEvent(CloudEvent cloudEvent) { * for the various triggers. CloudEvents are ones that follow the standards defined by cloudevents.io. * - * @param the type to be used in the {@link Unmarshallers} call when + * @param the type to be used in the {code Unmarshallers} call when * unmarshalling this event, if it is a CloudEvent. */ private abstract static class FunctionExecutor { @@ -320,20 +329,23 @@ void serviceCloudEvent(CloudEvent cloudEvent) throws Exception { /** Executes the user's background function. This can handle all HTTP methods. */ @Override - public void service(HttpServletRequest req, HttpServletResponse res) throws IOException { - String contentType = req.getContentType(); + public boolean handle(Request req, Response res, Callback callback) throws Exception { + String contentType = req.getHeaders().get(HttpHeader.CONTENT_TYPE); try { if ((contentType != null && contentType.startsWith("application/cloudevents+json")) - || req.getHeader("ce-specversion") != null) { + || req.getHeaders().get("ce-specversion") != null) { serviceCloudEvent(req); } else { serviceLegacyEvent(req); } - res.setStatus(HttpServletResponse.SC_OK); + res.setStatus(HttpStatus.OK_200); + callback.succeeded(); } catch (Throwable t) { - res.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); logger.log(Level.SEVERE, "Failed to execute " + functionExecutor.functionName(), t); + res.setStatus(HttpStatus.INTERNAL_SERVER_ERROR_500); + callback.succeeded(); } + return true; } private enum CloudEventKind { @@ -347,10 +359,11 @@ private enum CloudEventKind { * @param a fake type parameter, which corresponds to the type parameter of {@link * FunctionExecutor}. */ - private void serviceCloudEvent(HttpServletRequest req) throws Exception { + private void serviceCloudEvent(Request req) throws Exception { @SuppressWarnings("unchecked") FunctionExecutor executor = (FunctionExecutor) functionExecutor; - byte[] body = req.getInputStream().readAllBytes(); + + byte[] body = Content.Source.asByteArrayAsync(req, -1).get(); MessageReader reader = HttpMessageFactory.createReaderFromMultimap(headerMap(req), body); // It's important not to set the context ClassLoader earlier, because MessageUtils will use // ServiceLoader.load(EventFormat.class) to find a handler to deserialize a binary CloudEvent @@ -364,17 +377,16 @@ private void serviceCloudEvent(HttpServletRequest req) throws Exce // https://github.com/cloudevents/sdk-java/pull/259. } - private static Map> headerMap(HttpServletRequest req) { + private static Map> headerMap(Request req) { Map> headerMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); - for (String header : Collections.list(req.getHeaderNames())) { - for (String value : Collections.list(req.getHeaders(header))) { - headerMap.computeIfAbsent(header, unused -> new ArrayList<>()).add(value); - } + for (HttpField field : req.getHeaders()) { + headerMap.computeIfAbsent(field.getName(), unused -> new ArrayList<>()) + .addAll(field.getValueList()); } return headerMap; } - private void serviceLegacyEvent(HttpServletRequest req) throws Exception { + private void serviceLegacyEvent(Request req) throws Exception { Event event = parseLegacyEvent(req); runWithContextClassLoader(() -> functionExecutor.serviceLegacyEvent(event)); } diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java index 21115666..f51a23f3 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java @@ -19,12 +19,14 @@ import com.google.cloud.functions.invoker.http.HttpResponseImpl; import java.util.logging.Level; import java.util.logging.Logger; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpStatus; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.util.Callback; /** Executes the user's method. */ -public class HttpFunctionExecutor extends HttpServlet { +public class HttpFunctionExecutor extends Handler.Abstract { private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker"); private final HttpFunction function; @@ -59,19 +61,28 @@ public static HttpFunctionExecutor forClass(Class functionClass) { /** Executes the user's method, can handle all HTTP type methods. */ @Override - public void service(HttpServletRequest req, HttpServletResponse res) { - HttpRequestImpl reqImpl = new HttpRequestImpl(req); - HttpResponseImpl respImpl = new HttpResponseImpl(res); + public boolean handle(Request request, Response response, Callback callback) throws Exception { + + HttpRequestImpl reqImpl = new HttpRequestImpl(request); + HttpResponseImpl respImpl = new HttpResponseImpl(response); ClassLoader oldContextLoader = Thread.currentThread().getContextClassLoader(); try { Thread.currentThread().setContextClassLoader(function.getClass().getClassLoader()); function.service(reqImpl, respImpl); + respImpl.close(); + callback.succeeded(); } catch (Throwable t) { logger.log(Level.SEVERE, "Failed to execute " + function.getClass().getName(), t); - res.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + if (response.isCommitted()) { + callback.failed(t); + } else { + response.reset(); + response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR_500); + callback.succeeded(); + } } finally { Thread.currentThread().setContextClassLoader(oldContextLoader); - respImpl.flush(); } + return true; } } diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java index a6edfc32..45467a27 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java @@ -15,11 +15,13 @@ import java.util.Optional; import java.util.logging.Level; import java.util.logging.Logger; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpStatus; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.util.Callback; -public class TypedFunctionExecutor extends HttpServlet { +public class TypedFunctionExecutor extends Handler.Abstract { private static final String APPLY_METHOD = "apply"; private static final Logger logger = Logger.getLogger("com.google.cloud.functions.invoker"); @@ -94,7 +96,7 @@ static Optional handlerTypeArgument(Class> f /** Executes the user's method, can handle all HTTP type methods. */ @Override - public void service(HttpServletRequest req, HttpServletResponse res) { + public boolean handle(Request req, Response res, Callback callback) throws Exception { HttpRequestImpl reqImpl = new HttpRequestImpl(req); HttpResponseImpl resImpl = new HttpResponseImpl(res); ClassLoader oldContextClassLoader = Thread.currentThread().getContextClassLoader(); @@ -102,10 +104,20 @@ public void service(HttpServletRequest req, HttpServletResponse res) { try { Thread.currentThread().setContextClassLoader(function.getClass().getClassLoader()); handleRequest(reqImpl, resImpl); + resImpl.close(); + callback.succeeded(); + } catch (Throwable t) { + if (res.isCommitted()) { + callback.failed(t); + } else { + res.reset(); + res.setStatus(HttpStatus.INTERNAL_SERVER_ERROR_500); + callback.succeeded(); + } } finally { Thread.currentThread().setContextClassLoader(oldContextClassLoader); - resImpl.flush(); } + return true; } private void handleRequest(HttpRequest req, HttpResponse res) { @@ -114,7 +126,7 @@ private void handleRequest(HttpRequest req, HttpResponse res) { reqObj = format.deserialize(req, argType); } catch (Throwable t) { logger.log(Level.SEVERE, "Failed to parse request for " + function.getClass().getName(), t); - res.setStatusCode(HttpServletResponse.SC_BAD_REQUEST); + res.setStatusCode(HttpStatus.BAD_REQUEST_400); return; } @@ -123,7 +135,7 @@ private void handleRequest(HttpRequest req, HttpResponse res) { resObj = function.apply(reqObj); } catch (Throwable t) { logger.log(Level.SEVERE, "Failed to execute " + function.getClass().getName(), t); - res.setStatusCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + res.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR_500); return; } @@ -132,7 +144,7 @@ private void handleRequest(HttpRequest req, HttpResponse res) { } catch (Throwable t) { logger.log( Level.SEVERE, "Failed to serialize response for " + function.getClass().getName(), t); - res.setStatusCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + res.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR_500); return; } } @@ -147,7 +159,7 @@ private static class GsonWireFormat implements TypedFunction.WireFormat { @Override public void serialize(Object object, HttpResponse response) throws Exception { if (object == null) { - response.setStatusCode(HttpServletResponse.SC_NO_CONTENT); + response.setStatusCode(HttpStatus.NO_CONTENT_204); return; } try (BufferedWriter bodyWriter = response.getWriter()) { diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java index 2119645a..dc255776 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java @@ -21,26 +21,32 @@ import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; -import java.io.UncheckedIOException; -import java.util.AbstractMap.SimpleEntry; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.TreeMap; +import java.util.concurrent.ExecutionException; import java.util.regex.Matcher; import java.util.regex.Pattern; -import javax.servlet.ServletException; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.Part; +import java.util.stream.StreamSupport; +import org.eclipse.jetty.http.HttpField; +import org.eclipse.jetty.http.HttpFields; +import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.http.MultiPart; +import org.eclipse.jetty.http.MultiPart.Part; +import org.eclipse.jetty.http.MultiPartFormData; +import org.eclipse.jetty.io.Content; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.util.Fields.Field; public class HttpRequestImpl implements HttpRequest { - private final HttpServletRequest request; + private final Request request; - public HttpRequestImpl(HttpServletRequest request) { + public HttpRequestImpl(Request request) { this.request = request; } @@ -51,78 +57,105 @@ public String getMethod() { @Override public String getUri() { - String url = request.getRequestURL().toString(); - if (request.getQueryString() != null) { - url += "?" + request.getQueryString(); - } - return url; + return request.getHttpURI().asString(); } @Override public String getPath() { - return request.getRequestURI(); + return request.getHttpURI().getCanonicalPath(); } @Override public Optional getQuery() { - return Optional.ofNullable(request.getQueryString()); + return Optional.ofNullable(request.getHttpURI().getQuery()); } @Override public Map> getQueryParameters() { - return request.getParameterMap().entrySet().stream() - .collect(toMap(Map.Entry::getKey, e -> Arrays.asList(e.getValue()))); + + return Request.extractQueryParameters(request).stream() + .collect(toMap(Field::getName, Field::getValues)); } @Override public Map getParts() { - String contentType = request.getContentType(); - if (contentType == null || !request.getContentType().startsWith("multipart/form-data")) { + // TODO initiate reading the parts asynchronously before invocation + String contentType = request.getHeaders().get(HttpHeader.CONTENT_TYPE); + if (contentType == null || !contentType.startsWith("multipart/form-data")) { throw new IllegalStateException("Content-Type must be multipart/form-data: " + contentType); } + String boundary = MultiPart.extractBoundary(contentType); + if (boundary == null) { + throw new IllegalStateException("No boundary in content-type: " + contentType); + } try { - return request.getParts().stream().collect(toMap(Part::getName, HttpPartImpl::new)); - } catch (IOException e) { - throw new UncheckedIOException(e); - } catch (ServletException e) { - throw new RuntimeException(e.getMessage(), e); + MultiPartFormData.Parts parts = + MultiPartFormData.from(request, boundary, parser -> { + parser.setMaxMemoryFileSize(-1); + return parser.parse(request); + }).get(); + return StreamSupport.stream(parts.spliterator(), false) + .map(HttpPartImpl::new) + .collect(toMap(HttpPartImpl::getName, p -> p)); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); } } @Override public Optional getContentType() { - return Optional.ofNullable(request.getContentType()); + return Optional.ofNullable(request.getHeaders().get(HttpHeader.CONTENT_TYPE)); } @Override public long getContentLength() { - return request.getContentLength(); + return request.getLength(); } @Override public Optional getCharacterEncoding() { - return Optional.ofNullable(request.getCharacterEncoding()); + Charset charset = Request.getCharset(request); + return Optional.ofNullable(charset == null ? null : charset.name()); } + private InputStream inputStream; + private BufferedReader reader; + @Override public InputStream getInputStream() throws IOException { - return request.getInputStream(); + if (reader != null) { + throw new IllegalStateException("getReader() already called"); + } + if (inputStream == null) { + inputStream = Content.Source.asInputStream(request); + } + return inputStream; } @Override public BufferedReader getReader() throws IOException { - return request.getReader(); + if (reader == null) { + if (inputStream != null) { + throw new IllegalStateException("getInputStream already called"); + } + inputStream = Content.Source.asInputStream(request); + reader = new BufferedReader(new InputStreamReader(getInputStream(), + getCharacterEncoding().orElse(StandardCharsets.UTF_8.name()))); + } + return reader; } @Override public Map> getHeaders() { - return Collections.list(request.getHeaderNames()).stream() - .collect( - toMap( - name -> name, - name -> Collections.list(request.getHeaders(name)), - (a, b) -> b, - () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER))); + return toStringListMap(request.getHeaders()); + } + + static Map> toStringListMap(HttpFields headers) { + Map> map = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (HttpField field : headers) { + map.computeIfAbsent(field.getName(), key -> new ArrayList<>()).add(field.getValue()); + } + return map; } private static class HttpPartImpl implements HttpPart { @@ -132,19 +165,23 @@ private HttpPartImpl(Part part) { this.part = part; } + public String getName() { + return part.getName(); + } + @Override public Optional getFileName() { - return Optional.ofNullable(part.getSubmittedFileName()); + return Optional.ofNullable(part.getFileName()); } @Override public Optional getContentType() { - return Optional.ofNullable(part.getContentType()); + return Optional.ofNullable(part.getHeaders().get(HttpHeader.CONTENT_TYPE)); } @Override public long getContentLength() { - return part.getSize(); + return part.getLength(); } @Override @@ -160,7 +197,7 @@ public Optional getCharacterEncoding() { @Override public InputStream getInputStream() throws IOException { - return part.getInputStream(); + return Content.Source.asInputStream(part.newContentSource()); } @Override @@ -171,13 +208,16 @@ public BufferedReader getReader() throws IOException { @Override public Map> getHeaders() { - return part.getHeaderNames().stream() - .map(name -> new SimpleEntry<>(name, list(part.getHeaders(name)))) - .collect(toMap(Map.Entry::getKey, Map.Entry::getValue)); + return HttpRequestImpl.toStringListMap(part.getHeaders()); } private static List list(Collection collection) { return (collection instanceof List) ? (List) collection : new ArrayList<>(collection); } + + @Override + public String toString() { + return "%s{%s}".formatted(super.toString(), part); + } } } diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index c02246f0..e5a97196 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -14,24 +14,28 @@ package com.google.cloud.functions.invoker.http; -import static java.util.stream.Collectors.toMap; - import com.google.cloud.functions.HttpResponse; +import java.io.BufferedOutputStream; import java.io.BufferedWriter; import java.io.IOException; import java.io.OutputStream; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; -import java.util.TreeMap; -import javax.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.io.Content; +import org.eclipse.jetty.io.WriteThroughWriter; +import org.eclipse.jetty.server.Response; public class HttpResponseImpl implements HttpResponse { - private final HttpServletResponse response; + private final Response response; - public HttpResponseImpl(HttpServletResponse response) { + public HttpResponseImpl(Response response) { this.response = response; } @@ -43,75 +47,73 @@ public void setStatusCode(int code) { @Override @SuppressWarnings("deprecation") public void setStatusCode(int code, String message) { - response.setStatus(code, message); + response.setStatus(code); } @Override public void setContentType(String contentType) { - response.setContentType(contentType); + response.getHeaders().put(HttpHeader.CONTENT_TYPE, contentType); } @Override public Optional getContentType() { - return Optional.ofNullable(response.getContentType()); + return Optional.ofNullable(response.getHeaders().get(HttpHeader.CONTENT_TYPE)); } @Override public void appendHeader(String key, String value) { - response.addHeader(key, value); + response.getHeaders().add(key, value); } @Override public Map> getHeaders() { - return response.getHeaderNames().stream() - .collect( - toMap( - name -> name, - name -> new ArrayList<>(response.getHeaders(name)), - (a, b) -> b, - () -> new TreeMap<>(String.CASE_INSENSITIVE_ORDER))); + return HttpRequestImpl.toStringListMap(response.getHeaders()); } private static List list(Collection collection) { return (collection instanceof List) ? (List) collection : new ArrayList<>(collection); } + private OutputStream outputStream; + private BufferedWriter writer; + @Override - public OutputStream getOutputStream() throws IOException { - return response.getOutputStream(); + public OutputStream getOutputStream() { + if (writer != null) { + throw new IllegalStateException("getWriter called"); + } else if (outputStream == null) { + // TODO use BufferedSink when it is available + outputStream = new BufferedOutputStream(Content.Sink.asOutputStream(response)); + } + return outputStream; } - private BufferedWriter writer; - @Override public synchronized BufferedWriter getWriter() throws IOException { if (writer == null) { - // Unfortunately this means that we get two intermediate objects between the object we return - // and the underlying Writer that response.getWriter() wraps. We could try accessing the - // PrintWriter.out field via reflection, but that sort of access to non-public fields of - // platform classes is now frowned on and may draw warnings or even fail in subsequent - // versions. We could instead wrap the OutputStream, but that would require us to deduce the - // appropriate Charset, using logic like this: - // https://github.com/eclipse/jetty.project/blob/923ec38adf/jetty-server/src/main/java/org/eclipse/jetty/server/Response.java#L731 - // We may end up doing that if performance is an issue. - writer = new BufferedWriter(response.getWriter()); + if (outputStream != null) { + throw new IllegalStateException("getOutputStream called"); + } + String contentType = getContentType().orElse(null); + Charset charset = Objects.requireNonNullElse( + response.getRequest().getContext().getMimeTypes().getCharset(contentType), + StandardCharsets.UTF_8); + // TODO should we buffer in the input stream rather than as characters + outputStream = Content.Sink.asOutputStream(response); + writer = new BufferedWriter(WriteThroughWriter.newWriter(getOutputStream(), charset)); } return writer; } - public void flush() { + public void close() { try { - // We can't use HttpServletResponse.flushBuffer() because we wrap the - // PrintWriter returned by HttpServletResponse in our own BufferedWriter - // to match our API. So we have to flush whichever of getWriter() or - // getOutputStream() works. - try { - getOutputStream().flush(); - } catch (IllegalStateException e) { - getWriter().flush(); + if (writer != null) { + writer.close(); + } else if (outputStream != null) { + outputStream.close(); } } catch (IOException e) { - // Too bad, can't flush. + // Too bad, can't close. } } } diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java index ebc70718..943b7db8 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java @@ -44,23 +44,19 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; -import java.util.logging.Handler; import java.util.logging.Level; import java.util.logging.Logger; import java.util.stream.Stream; -import javax.servlet.MultipartConfigElement; -import javax.servlet.ServletException; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; import org.eclipse.jetty.http.HttpStatus; import org.eclipse.jetty.server.Connector; +import org.eclipse.jetty.server.Handler; import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; -import org.eclipse.jetty.server.handler.HandlerWrapper; -import org.eclipse.jetty.servlet.ServletContextHandler; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.server.handler.ContextHandler; +import org.eclipse.jetty.server.handler.ErrorHandler; +import org.eclipse.jetty.util.Callback; import org.eclipse.jetty.util.thread.QueuedThreadPool; /** @@ -87,7 +83,7 @@ public class Invoker { // if we arrange for them to be formatted using StackDriver's "structured // logging" JSON format. Remove the JDK's standard logger and replace it with // the JSON one. - for (Handler handler : rootLogger.getHandlers()) { + for (java.util.logging.Handler handler : rootLogger.getHandlers()) { rootLogger.removeHandler(handler); } rootLogger.addHandler(new JsonLogHandler(System.out, false)); @@ -283,34 +279,43 @@ private void startServer(boolean join) throws Exception { QueuedThreadPool pool = new QueuedThreadPool(1024); server = new Server(pool); + server.setErrorHandler(new ErrorHandler() { + @Override + public boolean handle(Request request, Response response, Callback callback) { + // Suppress error body + callback.succeeded(); + return true; + } + }); ServerConnector connector = new ServerConnector(server); connector.setPort(port); + connector.setReuseAddress(true); + connector.setReusePort(true); server.setConnectors(new Connector[] {connector}); - ServletContextHandler servletContextHandler = new ServletContextHandler(); - servletContextHandler.setContextPath("/"); - server.setHandler(NotFoundHandler.forServlet(servletContextHandler)); + ContextHandler contextHandler = new ContextHandler("/"); + server.setHandler(NotFoundHandler.forServlet(contextHandler)); Class functionClass = loadFunctionClass(); - HttpServlet servlet; + Handler handler; if (functionSignatureType == null) { - servlet = servletForDeducedSignatureType(functionClass); + handler = servletForDeducedSignatureType(functionClass); } else { switch (functionSignatureType) { case "http": if (TypedFunction.class.isAssignableFrom(functionClass)) { - servlet = TypedFunctionExecutor.forClass(functionClass); + handler = TypedFunctionExecutor.forClass(functionClass); } else { - servlet = HttpFunctionExecutor.forClass(functionClass); + handler = HttpFunctionExecutor.forClass(functionClass); } break; case "event": case "cloudevent": - servlet = BackgroundFunctionExecutor.forClass(functionClass); + handler = BackgroundFunctionExecutor.forClass(functionClass); break; case "typed": - servlet = TypedFunctionExecutor.forClass(functionClass); + handler = TypedFunctionExecutor.forClass(functionClass); break; default: String error = @@ -321,10 +326,10 @@ private void startServer(boolean join) throws Exception { throw new RuntimeException(error); } } - ServletHolder servletHolder = new ServletHolder(servlet); - servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("")); - servletContextHandler.addServlet(servletHolder, "/*"); + // TODO servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("")); + + server.getTail().setHandler(handler); server.start(); logServerInfo(); if (join) { @@ -371,7 +376,7 @@ private Class loadFunctionClass() throws ClassNotFoundException { } } - private HttpServlet servletForDeducedSignatureType(Class functionClass) { + private Handler servletForDeducedSignatureType(Class functionClass) { if (HttpFunction.class.isAssignableFrom(functionClass)) { return HttpFunctionExecutor.forClass(functionClass); } @@ -455,8 +460,8 @@ private static boolean isGcf() { * URL, meaning that someone testing their function by using a browser as an HTTP client can see * two requests, one for {@code /favicon.ico} and one for {@code /} (or whatever). */ - private static class NotFoundHandler extends HandlerWrapper { - static NotFoundHandler forServlet(ServletContextHandler servletHandler) { + private static class NotFoundHandler extends Handler.Wrapper { + static NotFoundHandler forServlet(ContextHandler servletHandler) { NotFoundHandler handler = new NotFoundHandler(); handler.setHandler(servletHandler); return handler; @@ -466,16 +471,14 @@ static NotFoundHandler forServlet(ServletContextHandler servletHandler) { new HashSet<>(Arrays.asList("/favicon.ico", "/robots.txt")); @Override - public void handle( - String target, - Request baseRequest, - HttpServletRequest request, - HttpServletResponse response) - throws IOException, ServletException { - if (NOT_FOUND_PATHS.contains(request.getRequestURI())) { - response.sendError(HttpStatus.NOT_FOUND_404, "Not Found"); + public boolean handle(Request request, Response response, Callback callback) throws Exception { + if (NOT_FOUND_PATHS.contains(request.getHttpURI().getCanonicalPath())) { + response.setStatus(HttpStatus.NOT_FOUND_404); + callback.succeeded(); + return true; } - super.handle(target, baseRequest, request, response); + + return super.handle(request, response, callback); } } diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java index f84ddbdd..a4b3300d 100644 --- a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java @@ -44,6 +44,7 @@ import java.net.URI; import java.net.URL; import java.net.URLEncoder; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; @@ -62,16 +63,17 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; +import org.eclipse.jetty.client.ByteBufferRequestContent; +import org.eclipse.jetty.client.ContentResponse; import org.eclipse.jetty.client.HttpClient; -import org.eclipse.jetty.client.api.ContentProvider; -import org.eclipse.jetty.client.api.ContentResponse; -import org.eclipse.jetty.client.api.Request; -import org.eclipse.jetty.client.util.BytesContentProvider; -import org.eclipse.jetty.client.util.MultiPartContentProvider; -import org.eclipse.jetty.client.util.StringContentProvider; +import org.eclipse.jetty.client.MultiPartRequestContent; +import org.eclipse.jetty.client.Request; +import org.eclipse.jetty.client.StringRequestContent; import org.eclipse.jetty.http.HttpFields; import org.eclipse.jetty.http.HttpHeader; import org.eclipse.jetty.http.HttpStatus; +import org.eclipse.jetty.http.MultiPart; +import org.eclipse.jetty.http.MultiPart.ContentSourcePart; import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; @@ -161,7 +163,7 @@ abstract static class TestCase { abstract String url(); - abstract ContentProvider requestContent(); + abstract Request.Content requestContent(); abstract int expectedResponseCode(); @@ -194,10 +196,10 @@ abstract static class Builder { abstract Builder setUrl(String x); - abstract Builder setRequestContent(ContentProvider x); + abstract Builder setRequestContent(Request.Content x); Builder setRequestText(String text) { - return setRequestContent(new StringContentProvider(text)); + return setRequestContent(new StringRequestContent(text)); } abstract Builder setExpectedResponseCode(int x); @@ -257,7 +259,7 @@ public void exceptionHttp() throws Exception { String exceptionExpectedOutput = "\"severity\": \"ERROR\", \"logging.googleapis.com/sourceLocation\": {\"file\":" + " \"com/google/cloud/functions/invoker/HttpFunctionExecutor.java\", \"method\":" - + " \"service\"}, \"message\": \"Failed to execute" + + " \"handle\"}, \"message\": \"Failed to execute" + " com.google.cloud.functions.invoker.testfunctions.ExceptionHttp\\n" + "java.lang.RuntimeException: exception thrown for test"; testHttpFunction( @@ -274,7 +276,7 @@ public void exceptionBackground() throws Exception { String exceptionExpectedOutput = "\"severity\": \"ERROR\", \"logging.googleapis.com/sourceLocation\": {\"file\":" + " \"com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java\", \"method\":" - + " \"service\"}, \"message\": \"Failed to execute" + + " \"handle\"}, \"message\": \"Failed to execute" + " com.google.cloud.functions.invoker.testfunctions.ExceptionBackground\\n" + "java.lang.RuntimeException: exception thrown for test"; @@ -533,11 +535,14 @@ public void packageless() throws Exception { @Test public void multipart() throws Exception { - MultiPartContentProvider multiPartProvider = new MultiPartContentProvider(); + MultiPartRequestContent multiPartRequestContent = new MultiPartRequestContent(); byte[] bytes = new byte[17]; - multiPartProvider.addFieldPart("bytes", new BytesContentProvider(bytes), new HttpFields()); - String string = "1234567890"; - multiPartProvider.addFieldPart("string", new StringContentProvider(string), new HttpFields()); + multiPartRequestContent.addPart(new ContentSourcePart("bytes", null, + HttpFields.EMPTY, new ByteBufferRequestContent(ByteBuffer.wrap(bytes)))); + multiPartRequestContent.addPart(new MultiPart.ContentSourcePart("string", null, + HttpFields.EMPTY, new StringRequestContent("1234567890"))); + multiPartRequestContent.close(); + String expectedResponse = "part bytes type application/octet-stream length 17\n" + "part string type text/plain;charset=UTF-8 length 10\n"; @@ -545,8 +550,8 @@ public void multipart() throws Exception { fullTarget("Multipart"), ImmutableList.of( TestCase.builder() - .setHttpContentType(Optional.empty()) - .setRequestContent(multiPartProvider) + .setHttpContentType(multiPartRequestContent.getContentType()) + .setRequestContent(multiPartRequestContent) .setExpectedResponseText(expectedResponse) .build())); } @@ -680,11 +685,12 @@ private void testFunction( testCase.snoopFile().ifPresent(File::delete); String uri = "http://localhost:" + serverPort + testCase.url(); Request request = httpClient.POST(uri); - testCase - .httpContentType() - .ifPresent(contentType -> request.header(HttpHeader.CONTENT_TYPE, contentType)); - testCase.httpHeaders().forEach((header, value) -> request.header(header, value)); - request.content(testCase.requestContent()); + + request.headers(m -> { + testCase.httpContentType().ifPresent(contentType -> m.put(HttpHeader.CONTENT_TYPE, contentType)); + testCase.httpHeaders().forEach(m::put); + }); + request.body(testCase.requestContent()); ContentResponse response = request.send(); expect .withMessage("Response to %s is %s %s", uri, response.getStatus(), response.getReason()) diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java index e52ec62a..ad3789d5 100644 --- a/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java @@ -28,6 +28,7 @@ import java.io.InputStreamReader; import java.io.OutputStream; import java.net.ServerSocket; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -35,22 +36,22 @@ import java.util.TreeMap; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; -import javax.servlet.MultipartConfigElement; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; +import org.eclipse.jetty.client.ByteBufferRequestContent; +import org.eclipse.jetty.client.ContentResponse; import org.eclipse.jetty.client.HttpClient; -import org.eclipse.jetty.client.api.ContentResponse; -import org.eclipse.jetty.client.api.Request; -import org.eclipse.jetty.client.util.BytesContentProvider; -import org.eclipse.jetty.client.util.MultiPartContentProvider; -import org.eclipse.jetty.client.util.StringContentProvider; +import org.eclipse.jetty.client.MultiPartRequestContent; +import org.eclipse.jetty.client.StringRequestContent; import org.eclipse.jetty.http.HttpFields; import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.http.HttpMethod; import org.eclipse.jetty.http.HttpStatus; +import org.eclipse.jetty.http.HttpStatus.Code; +import org.eclipse.jetty.http.MultiPart; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; import org.eclipse.jetty.server.Server; -import org.eclipse.jetty.servlet.ServletContextHandler; -import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.util.Callback; import org.junit.BeforeClass; import org.junit.Test; @@ -82,21 +83,17 @@ public static void allocateServerPort() throws IOException { } /** - * Wrapper class that allows us to start a Jetty server with a single servlet for {@code /*} - * within a try-with-resources statement. The servlet will be configured to support multipart + * Wrapper class that allows us to start a Jetty server with a single handler for {@code /*} + * within a try-with-resources statement. The handler will be configured to support multipart * requests. */ private static class SimpleServer implements AutoCloseable { private final Server server; - SimpleServer(HttpServlet servlet) throws Exception { + SimpleServer(Handler handler) throws Exception { this.server = new Server(serverPort); - ServletContextHandler context = new ServletContextHandler(); - context.setContextPath("/"); - server.setHandler(context); - ServletHolder servletHolder = new ServletHolder(servlet); - servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("tiddly")); - context.addServlet(servletHolder, "/*"); + server.setHandler(handler); + // TODO servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("tiddly")); server.start(); } @@ -113,16 +110,16 @@ private interface HttpRequestTest { /** * Tests methods on the {@link HttpRequest} object while the request is being serviced. We are not - * guaranteed that the underlying {@link HttpServletRequest} object will still be valid when the + * guaranteed that the underlying {@link Request} object will still be valid when the * request completes, and in fact in Jetty it isn't. So we perform the checks in the context of - * the servlet, and report any exception back to the test method. + * the handler, and report any exception back to the test method. */ @Test public void httpRequestMethods() throws Exception { AtomicReference testReference = new AtomicReference<>(); AtomicReference exceptionReference = new AtomicReference<>(); - HttpRequestServlet testServlet = new HttpRequestServlet(testReference, exceptionReference); - try (SimpleServer server = new SimpleServer(testServlet)) { + HttpRequestHandler testHandler = new HttpRequestHandler(testReference, exceptionReference); + try (SimpleServer server = new SimpleServer(testHandler)) { httpRequestMethods(testReference, exceptionReference); } } @@ -193,14 +190,16 @@ private void httpRequestMethods( }; for (HttpRequestTest test : tests) { testReference.set(test); - Request request = + org.eclipse.jetty.client.Request request = httpClient .POST(uri) - .header(HttpHeader.CONTENT_TYPE, "text/plain; charset=utf-8") - .header("foo", "bar") - .header("foo", "baz") - .header("CaSe-SeNsItIvE", "VaLuE") - .content(new StringContentProvider(TEST_BODY)); + .headers(m -> { + m.add(HttpHeader.CONTENT_TYPE, "text/plain; charset=utf-8"); + m.add("foo", "bar"); + m.add("foo", "baz"); + m.add("CaSe-SeNsItIvE", "VaLuE"); + }) + .body(new StringRequestContent(TEST_BODY)); ContentResponse response = request.send(); assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); throwIfNotNull(exceptionReference.get()); @@ -223,8 +222,8 @@ public void emptyRequest() throws Exception { }; AtomicReference exceptionReference = new AtomicReference<>(); AtomicReference testReference = new AtomicReference<>(test); - HttpRequestServlet testServlet = new HttpRequestServlet(testReference, exceptionReference); - try (SimpleServer server = new SimpleServer(testServlet)) { + HttpRequestHandler testHandler = new HttpRequestHandler(testReference, exceptionReference); + try (SimpleServer server = new SimpleServer(testHandler)) { ContentResponse response = httpClient.POST(uri).send(); assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); throwIfNotNull(exceptionReference.get()); @@ -240,20 +239,22 @@ private void validateReader(BufferedReader reader) { public void multiPartRequest() throws Exception { AtomicReference testReference = new AtomicReference<>(); AtomicReference exceptionReference = new AtomicReference<>(); - HttpRequestServlet testServlet = new HttpRequestServlet(testReference, exceptionReference); + HttpRequestHandler testHandler = new HttpRequestHandler(testReference, exceptionReference); HttpClient httpClient = new HttpClient(); httpClient.start(); String uri = "http://localhost:" + serverPort + "/"; - MultiPartContentProvider multiPart = new MultiPartContentProvider(); - HttpFields textHttpFields = new HttpFields(); - textHttpFields.add("foo", "bar"); - multiPart.addFieldPart("text", new StringContentProvider(TEST_BODY), textHttpFields); - HttpFields bytesHttpFields = new HttpFields(); - bytesHttpFields.add("foo", "baz"); - bytesHttpFields.add("foo", "buh"); + MultiPartRequestContent multiPart = new MultiPartRequestContent(); + HttpFields textHttpFields = HttpFields.build() + .add("foo", "bar"); + multiPart.addPart(new MultiPart.ContentSourcePart("text", null, textHttpFields, + new StringRequestContent(TEST_BODY))); + HttpFields.Mutable bytesHttpFields = HttpFields.build() + .add("foo", "baz") + .add("foo", "buh"); assertThat(bytesHttpFields.getValuesList("foo")).containsExactly("baz", "buh"); - multiPart.addFilePart( - "binary", "/tmp/binary.x", new BytesContentProvider(RANDOM_BYTES), bytesHttpFields); + multiPart.addPart(new MultiPart.ContentSourcePart("binary", "/tmp/binary.x", + bytesHttpFields, new ByteBufferRequestContent(ByteBuffer.wrap(RANDOM_BYTES)))); + multiPart.close(); HttpRequestTest test = request -> { // The Content-Type header will also have a boundary=something attribute. @@ -272,10 +273,9 @@ public void multiPartRequest() throws Exception { assertThat(bytesPart.getFileName()).hasValue("/tmp/binary.x"); assertThat(bytesPart.getContentLength()).isEqualTo(RANDOM_BYTES.length); assertThat(bytesPart.getContentType()).hasValue("application/octet-stream"); - // We only see ["buh"] here, not ["baz", "buh"], apparently due to a Jetty bug. - // Repeated headers on multi-part content are not a big problem anyway. List foos = bytesPart.getHeaders().get("foo"); - assertThat(foos).contains("buh"); + assertThat(foos).containsExactly("baz", "buh"); + byte[] bytes = new byte[RANDOM_BYTES.length]; try (InputStream inputStream = bytesPart.getInputStream()) { assertThat(inputStream.read(bytes)).isEqualTo(bytes.length); @@ -283,20 +283,22 @@ public void multiPartRequest() throws Exception { assertThat(bytes).isEqualTo(RANDOM_BYTES); } }; - try (SimpleServer server = new SimpleServer(testServlet)) { + try (SimpleServer server = new SimpleServer(testHandler)) { testReference.set(test); - Request request = httpClient.POST(uri).header("foo", "oof").content(multiPart); + org.eclipse.jetty.client.Request request = httpClient.POST(uri) + .headers(m -> m.put("foo", "oof")) + .body(multiPart); ContentResponse response = request.send(); assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); throwIfNotNull(exceptionReference.get()); } } - private static class HttpRequestServlet extends HttpServlet { + private static class HttpRequestHandler extends Handler.Abstract { private final AtomicReference testReference; private final AtomicReference exceptionReference; - private HttpRequestServlet( + private HttpRequestHandler( AtomicReference testReference, AtomicReference exceptionReference) { this.testReference = testReference; @@ -304,12 +306,21 @@ private HttpRequestServlet( } @Override - protected void doPost(HttpServletRequest req, HttpServletResponse resp) { + public boolean handle(Request request, Response response, Callback callback) { try { - testReference.get().test(new HttpRequestImpl(req)); + if (!HttpMethod.POST.is(request.getMethod())) { + response.setStatus(HttpStatus.METHOD_NOT_ALLOWED_405); + callback.succeeded(); + return true; + } + + testReference.get().test(new HttpRequestImpl(request)); } catch (Throwable t) { + t.printStackTrace(); exceptionReference.set(t); } + callback.succeeded(); + return true; } } @@ -327,8 +338,8 @@ private interface HttpResponseTest { public void httpResponseSetAndGet() throws Exception { AtomicReference testReference = new AtomicReference<>(); AtomicReference exceptionReference = new AtomicReference<>(); - HttpResponseServlet testServlet = new HttpResponseServlet(testReference, exceptionReference); - try (SimpleServer server = new SimpleServer(testServlet)) { + HttpResponseHandler testHandler = new HttpResponseHandler(testReference, exceptionReference); + try (SimpleServer server = new SimpleServer(testHandler)) { httpResponseSetAndGet(testReference, exceptionReference); } } @@ -350,8 +361,7 @@ private void httpResponseSetAndGet( .containsAtLeast("Content-Type", Arrays.asList("application/octet-stream")); }, response -> { - Map> initialHeaders = response.getHeaders(); - // The servlet spec says this should be empty, but actually we get a Date header here. + // The fields are initialized with a Date header as per the HTTP RFCs. // So we just check that we can add our own headers. response.appendHeader("foo", "bar"); response.appendHeader("wibbly", "wobbly"); @@ -366,18 +376,18 @@ private void httpResponseSetAndGet( HttpClient httpClient = new HttpClient(); httpClient.start(); String uri = "http://localhost:" + serverPort; - Request request = httpClient.POST(uri); + org.eclipse.jetty.client.Request request = httpClient.POST(uri); ContentResponse response = request.send(); assertThat(response.getStatus()).isEqualTo(HttpStatus.OK_200); throwIfNotNull(exceptionReference.get()); } } - private static class HttpResponseServlet extends HttpServlet { + private static class HttpResponseHandler extends Handler.Abstract { private final AtomicReference testReference; private final AtomicReference exceptionReference; - private HttpResponseServlet( + private HttpResponseHandler( AtomicReference testReference, AtomicReference exceptionReference) { this.testReference = testReference; @@ -385,12 +395,18 @@ private HttpResponseServlet( } @Override - protected void doPost(HttpServletRequest req, HttpServletResponse resp) { + public boolean handle(Request request, Response response, Callback callback) { + if (!HttpMethod.POST.is(request.getMethod())) { + return false; + } try { - testReference.get().test(new HttpResponseImpl(resp)); + testReference.get().test(new HttpResponseImpl(response)); + callback.succeeded(); } catch (Throwable t) { exceptionReference.set(t); + Response.writeError(request, response, callback, t); } + return true; } } @@ -417,15 +433,15 @@ private static ResponseTest responseTest( /** * Tests that operations on the {@link HttpResponse} have the appropriate effect on the HTTP * response that ends up being sent. Here, for each check, we have two operations: the operation - * on the {@link HttpResponse}, which happens inside the servlet, and the operation to check the + * on the {@link HttpResponse}, which happens inside the handler, and the operation to check the * HTTP result, which happens in the client thread. */ @Test public void httpResponseEffects() throws Exception { AtomicReference testReference = new AtomicReference<>(); AtomicReference exceptionReference = new AtomicReference<>(); - HttpResponseServlet testServlet = new HttpResponseServlet(testReference, exceptionReference); - try (SimpleServer server = new SimpleServer(testServlet)) { + HttpResponseHandler testHandler = new HttpResponseHandler(testReference, exceptionReference); + try (SimpleServer server = new SimpleServer(testHandler)) { httpResponseEffects(testReference, exceptionReference); } } @@ -445,10 +461,11 @@ private void httpResponseEffects( response -> response.setStatusCode(HttpStatus.IM_A_TEAPOT_418), response -> assertThat(response.getStatus()).isEqualTo(HttpStatus.IM_A_TEAPOT_418)), responseTest( + // reason string cannot be set by application response -> response.setStatusCode(HttpStatus.IM_A_TEAPOT_418, "Je suis une théière"), response -> { assertThat(response.getStatus()).isEqualTo(HttpStatus.IM_A_TEAPOT_418); - assertThat(response.getReason()).isEqualTo("Je suis une théière"); + assertThat(response.getReason()).isEqualTo(Code.IM_A_TEAPOT.getMessage()); }), responseTest( response -> response.setContentType("application/noddy"), @@ -491,7 +508,7 @@ private void httpResponseEffects( HttpClient httpClient = new HttpClient(); httpClient.start(); String uri = "http://localhost:" + serverPort; - Request request = httpClient.POST(uri); + org.eclipse.jetty.client.Request request = httpClient.POST(uri); ContentResponse response = request.send(); throwIfNotNull(exceptionReference.get()); test.responseCheck.test(response); From e8e2d4d3167e44d6950d033982a3fb1d13aede43 Mon Sep 17 00:00:00 2001 From: gregw Date: Tue, 26 Sep 2023 14:36:19 +1000 Subject: [PATCH 2/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../invoker/http/HttpRequestImpl.java | 33 ++++++++++++------- .../invoker/http/HttpResponseImpl.java | 11 ++----- .../functions/invoker/runner/Invoker.java | 23 ++++--------- .../functions/invoker/http/HttpTest.java | 1 - 4 files changed, 29 insertions(+), 39 deletions(-) diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java index dc255776..ed7d4cb3 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java @@ -14,8 +14,6 @@ package com.google.cloud.functions.invoker.http; -import static java.util.stream.Collectors.toMap; - import com.google.cloud.functions.HttpRequest; import java.io.BufferedReader; import java.io.IOException; @@ -25,6 +23,8 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -32,7 +32,6 @@ import java.util.concurrent.ExecutionException; import java.util.regex.Matcher; import java.util.regex.Pattern; -import java.util.stream.StreamSupport; import org.eclipse.jetty.http.HttpField; import org.eclipse.jetty.http.HttpFields; import org.eclipse.jetty.http.HttpHeader; @@ -41,10 +40,12 @@ import org.eclipse.jetty.http.MultiPartFormData; import org.eclipse.jetty.io.Content; import org.eclipse.jetty.server.Request; -import org.eclipse.jetty.util.Fields.Field; +import org.eclipse.jetty.util.Fields; public class HttpRequestImpl implements HttpRequest { private final Request request; + private InputStream inputStream; + private BufferedReader reader; public HttpRequestImpl(Request request) { this.request = request; @@ -72,9 +73,15 @@ public Optional getQuery() { @Override public Map> getQueryParameters() { + Fields fields = Request.extractQueryParameters(request); + if (fields.isEmpty()) { + return Collections.emptyMap(); + } - return Request.extractQueryParameters(request).stream() - .collect(toMap(Field::getName, Field::getValues)); + Map> map = new HashMap<>(); + fields.forEach(field -> map.put(field.getName(), + Collections.unmodifiableList(field.getValues()))); + return Collections.unmodifiableMap(map); } @Override @@ -94,9 +101,14 @@ public Map getParts() { parser.setMaxMemoryFileSize(-1); return parser.parse(request); }).get(); - return StreamSupport.stream(parts.spliterator(), false) - .map(HttpPartImpl::new) - .collect(toMap(HttpPartImpl::getName, p -> p)); + + if (parts.size() == 0) { + return Collections.emptyMap(); + } + + Map map = new HashMap<>(); + parts.forEach(part -> map.put(part.getName(), new HttpPartImpl(part))); + return Collections.unmodifiableMap(map); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } @@ -118,9 +130,6 @@ public Optional getCharacterEncoding() { return Optional.ofNullable(charset == null ? null : charset.name()); } - private InputStream inputStream; - private BufferedReader reader; - @Override public InputStream getInputStream() throws IOException { if (reader != null) { diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index e5a97196..c8b36efe 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -21,8 +21,6 @@ import java.io.OutputStream; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Objects; @@ -34,6 +32,8 @@ public class HttpResponseImpl implements HttpResponse { private final Response response; + private OutputStream outputStream; + private BufferedWriter writer; public HttpResponseImpl(Response response) { this.response = response; @@ -70,13 +70,6 @@ public Map> getHeaders() { return HttpRequestImpl.toStringListMap(response.getHeaders()); } - private static List list(Collection collection) { - return (collection instanceof List) ? (List) collection : new ArrayList<>(collection); - } - - private OutputStream outputStream; - private BufferedWriter writer; - @Override public OutputStream getOutputStream() { if (writer != null) { diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java index 943b7db8..cd58302b 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java @@ -54,7 +54,6 @@ import org.eclipse.jetty.server.Response; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.ServerConnector; -import org.eclipse.jetty.server.handler.ContextHandler; import org.eclipse.jetty.server.handler.ErrorHandler; import org.eclipse.jetty.util.Callback; import org.eclipse.jetty.util.thread.QueuedThreadPool; @@ -234,7 +233,7 @@ ClassLoader getFunctionClassLoader() { * unit or integration test, use {@link #startTestServer()} instead. * * @see #stopServer() - * @throws Exception + * @throws Exception If there was a problem starting the server */ public void startServer() throws Exception { startServer(true); @@ -266,7 +265,7 @@ public void startServer() throws Exception { * } * * @see #stopServer() - * @throws Exception + * @throws Exception If there was a problem starting the server */ public void startTestServer() throws Exception { startServer(false); @@ -292,15 +291,13 @@ public boolean handle(Request request, Response response, Callback callback) { connector.setReuseAddress(true); connector.setReusePort(true); server.setConnectors(new Connector[] {connector}); - - ContextHandler contextHandler = new ContextHandler("/"); - server.setHandler(NotFoundHandler.forServlet(contextHandler)); + server.setHandler(new NotFoundHandler()); Class functionClass = loadFunctionClass(); Handler handler; if (functionSignatureType == null) { - handler = servletForDeducedSignatureType(functionClass); + handler = handlerForDeducedSignatureType(functionClass); } else { switch (functionSignatureType) { case "http": @@ -327,8 +324,6 @@ public boolean handle(Request request, Response response, Callback callback) { } } - // TODO servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("")); - server.getTail().setHandler(handler); server.start(); logServerInfo(); @@ -376,7 +371,7 @@ private Class loadFunctionClass() throws ClassNotFoundException { } } - private Handler servletForDeducedSignatureType(Class functionClass) { + private Handler handlerForDeducedSignatureType(Class functionClass) { if (HttpFunction.class.isAssignableFrom(functionClass)) { return HttpFunctionExecutor.forClass(functionClass); } @@ -456,16 +451,11 @@ private static boolean isGcf() { /** * Wrapper that intercepts requests for {@code /favicon.ico} and {@code /robots.txt} and causes - * them to produce a 404 status. Otherwise they would be sent to the function code, like any other + * them to produce a 404 status. Otherwise, they would be sent to the function code, like any other * URL, meaning that someone testing their function by using a browser as an HTTP client can see * two requests, one for {@code /favicon.ico} and one for {@code /} (or whatever). */ private static class NotFoundHandler extends Handler.Wrapper { - static NotFoundHandler forServlet(ContextHandler servletHandler) { - NotFoundHandler handler = new NotFoundHandler(); - handler.setHandler(servletHandler); - return handler; - } private static final Set NOT_FOUND_PATHS = new HashSet<>(Arrays.asList("/favicon.ico", "/robots.txt")); @@ -507,7 +497,6 @@ private static class OnlyApiClassLoader extends ClassLoader { protected Class findClass(String name) throws ClassNotFoundException { String prefix = "com.google.cloud.functions."; if ((name.startsWith(prefix) && Character.isUpperCase(name.charAt(prefix.length()))) - || name.startsWith("javax.servlet.") || isCloudEventsApiClass(name)) { return runtimeClassLoader.loadClass(name); } diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java index ad3789d5..20cdbe56 100644 --- a/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/http/HttpTest.java @@ -93,7 +93,6 @@ private static class SimpleServer implements AutoCloseable { SimpleServer(Handler handler) throws Exception { this.server = new Server(serverPort); server.setHandler(handler); - // TODO servletHolder.getRegistration().setMultipartConfig(new MultipartConfigElement("tiddly")); server.start(); } From 5417e87f64293c0ac74f88839cc13945f8cca96e Mon Sep 17 00:00:00 2001 From: gregw Date: Tue, 26 Sep 2023 16:37:42 +1000 Subject: [PATCH 3/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. Several other minor optimizations are included. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../invoker/http/HttpRequestImpl.java | 29 +++++++------------ .../invoker/http/HttpResponseImpl.java | 17 ++++++----- 2 files changed, 21 insertions(+), 25 deletions(-) diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java index ed7d4cb3..774e8e16 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java @@ -22,19 +22,18 @@ import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.TreeMap; import java.util.concurrent.ExecutionException; -import java.util.regex.Matcher; -import java.util.regex.Pattern; import org.eclipse.jetty.http.HttpField; import org.eclipse.jetty.http.HttpFields; import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.http.MimeTypes; import org.eclipse.jetty.http.MultiPart; import org.eclipse.jetty.http.MultiPart.Part; import org.eclipse.jetty.http.MultiPartFormData; @@ -149,7 +148,7 @@ public BufferedReader getReader() throws IOException { } inputStream = Content.Source.asInputStream(request); reader = new BufferedReader(new InputStreamReader(getInputStream(), - getCharacterEncoding().orElse(StandardCharsets.UTF_8.name()))); + Objects.requireNonNullElse(Request.getCharset(request), StandardCharsets.UTF_8))); } return reader; } @@ -169,9 +168,11 @@ static Map> toStringListMap(HttpFields headers) { private static class HttpPartImpl implements HttpPart { private final Part part; + private final String contentType; private HttpPartImpl(Part part) { this.part = part; + contentType = part.getHeaders().get(HttpHeader.CONTENT_TYPE); } public String getName() { @@ -185,7 +186,7 @@ public Optional getFileName() { @Override public Optional getContentType() { - return Optional.ofNullable(part.getHeaders().get(HttpHeader.CONTENT_TYPE)); + return Optional.ofNullable(contentType); } @Override @@ -195,13 +196,7 @@ public long getContentLength() { @Override public Optional getCharacterEncoding() { - String contentType = getContentType().orElse(null); - if (contentType == null) { - return Optional.empty(); - } - Pattern charsetPattern = Pattern.compile("(?i).*;\\s*charset\\s*=([^;\\s]*)\\s*(;|$)"); - Matcher matcher = charsetPattern.matcher(contentType); - return matcher.matches() ? Optional.of(matcher.group(1)) : Optional.empty(); + return Optional.ofNullable(MimeTypes.getCharsetFromContentType(contentType)); } @Override @@ -211,8 +206,10 @@ public InputStream getInputStream() throws IOException { @Override public BufferedReader getReader() throws IOException { - String encoding = getCharacterEncoding().orElse("utf-8"); - return new BufferedReader(new InputStreamReader(getInputStream(), encoding)); + return new BufferedReader( + new InputStreamReader(getInputStream(), + Objects.requireNonNullElse(MimeTypes.DEFAULTS.getCharset(contentType), + StandardCharsets.UTF_8))); } @Override @@ -220,10 +217,6 @@ public Map> getHeaders() { return HttpRequestImpl.toStringListMap(part.getHeaders()); } - private static List list(Collection collection) { - return (collection instanceof List) ? (List) collection : new ArrayList<>(collection); - } - @Override public String toString() { return "%s{%s}".formatted(super.toString(), part); diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index c8b36efe..3a59cb6e 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -34,6 +34,7 @@ public class HttpResponseImpl implements HttpResponse { private final Response response; private OutputStream outputStream; private BufferedWriter writer; + private Charset charset; public HttpResponseImpl(Response response) { this.response = response; @@ -53,6 +54,7 @@ public void setStatusCode(int code, String message) { @Override public void setContentType(String contentType) { response.getHeaders().put(HttpHeader.CONTENT_TYPE, contentType); + charset = response.getRequest().getContext().getMimeTypes().getCharset(contentType); } @Override @@ -62,7 +64,11 @@ public Optional getContentType() { @Override public void appendHeader(String key, String value) { - response.getHeaders().add(key, value); + if (HttpHeader.CONTENT_TYPE.is(key)) { + setContentType(value); + } else { + response.getHeaders().add(key, value); + } } @Override @@ -87,13 +93,10 @@ public synchronized BufferedWriter getWriter() throws IOException { if (outputStream != null) { throw new IllegalStateException("getOutputStream called"); } - String contentType = getContentType().orElse(null); - Charset charset = Objects.requireNonNullElse( - response.getRequest().getContext().getMimeTypes().getCharset(contentType), - StandardCharsets.UTF_8); - // TODO should we buffer in the input stream rather than as characters outputStream = Content.Sink.asOutputStream(response); - writer = new BufferedWriter(WriteThroughWriter.newWriter(getOutputStream(), charset)); + // TODO extend close in such a way as to make the buffer flush the last write. + writer = new BufferedWriter(WriteThroughWriter.newWriter(getOutputStream(), + Objects.requireNonNullElse(charset, StandardCharsets.UTF_8))); } return writer; } From f292929ccebae4c1b8d4b64f5081eefb056907d2 Mon Sep 17 00:00:00 2001 From: gregw Date: Tue, 26 Sep 2023 19:40:42 +1000 Subject: [PATCH 4/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. Several optimizations are included. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../invoker/HttpFunctionExecutor.java | 3 +- .../invoker/TypedFunctionExecutor.java | 3 +- .../invoker/http/HttpResponseImpl.java | 118 +++++++++++++++--- 3 files changed, 106 insertions(+), 18 deletions(-) diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java index f51a23f3..c97c3d4a 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/HttpFunctionExecutor.java @@ -69,8 +69,7 @@ public boolean handle(Request request, Response response, Callback callback) thr try { Thread.currentThread().setContextClassLoader(function.getClass().getClassLoader()); function.service(reqImpl, respImpl); - respImpl.close(); - callback.succeeded(); + respImpl.close(callback); } catch (Throwable t) { logger.log(Level.SEVERE, "Failed to execute " + function.getClass().getName(), t); if (response.isCommitted()) { diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java index 45467a27..24dd71d3 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/TypedFunctionExecutor.java @@ -104,8 +104,7 @@ public boolean handle(Request req, Response res, Callback callback) throws Excep try { Thread.currentThread().setContextClassLoader(function.getClass().getClassLoader()); handleRequest(reqImpl, resImpl); - resImpl.close(); - callback.succeeded(); + resImpl.close(callback); } catch (Throwable t) { if (res.isCommitted()) { callback.failed(t); diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index 3a59cb6e..81d8c5cf 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -15,10 +15,10 @@ package com.google.cloud.functions.invoker.http; import com.google.cloud.functions.HttpResponse; -import java.io.BufferedOutputStream; import java.io.BufferedWriter; import java.io.IOException; import java.io.OutputStream; +import java.io.Writer; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.List; @@ -28,11 +28,15 @@ import org.eclipse.jetty.http.HttpHeader; import org.eclipse.jetty.io.Content; import org.eclipse.jetty.io.WriteThroughWriter; +import org.eclipse.jetty.io.content.BufferedContentSink; +import org.eclipse.jetty.io.content.ContentSinkOutputStream; +import org.eclipse.jetty.server.Request; import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.util.Callback; public class HttpResponseImpl implements HttpResponse { private final Response response; - private OutputStream outputStream; + private ContentSinkOutputStream contentSinkOutputStream; private BufferedWriter writer; private Charset charset; @@ -80,36 +84,122 @@ public Map> getHeaders() { public OutputStream getOutputStream() { if (writer != null) { throw new IllegalStateException("getWriter called"); - } else if (outputStream == null) { - // TODO use BufferedSink when it is available - outputStream = new BufferedOutputStream(Content.Sink.asOutputStream(response)); + } else if (contentSinkOutputStream == null) { + Request request = response.getRequest(); + int outputBufferSize = request.getConnectionMetaData().getHttpConfiguration() + .getOutputBufferSize(); + BufferedContentSink bufferedContentSink = new BufferedContentSink(response, + request.getComponents().getByteBufferPool(), + false, outputBufferSize / 2, outputBufferSize); + contentSinkOutputStream = new ContentSinkOutputStream(bufferedContentSink); } - return outputStream; + return contentSinkOutputStream; } @Override public synchronized BufferedWriter getWriter() throws IOException { if (writer == null) { - if (outputStream != null) { + if (contentSinkOutputStream != null) { throw new IllegalStateException("getOutputStream called"); } - outputStream = Content.Sink.asOutputStream(response); - // TODO extend close in such a way as to make the buffer flush the last write. - writer = new BufferedWriter(WriteThroughWriter.newWriter(getOutputStream(), + + writer = new NonBufferedWriter(WriteThroughWriter.newWriter(getOutputStream(), Objects.requireNonNullElse(charset, StandardCharsets.UTF_8))); } return writer; } - public void close() { + /** + * Close the response, flushing all content. + * + * @param callback a {@link Callback} to be completed when the response is closed. + */ + public void close(Callback callback) { try { if (writer != null) { - writer.close(); - } else if (outputStream != null) { - outputStream.close(); + writer.flush(); + } + if (contentSinkOutputStream != null) { + // Do an asynchronous close, so large buffered content may be written without blocking + contentSinkOutputStream.close(callback); + } else { + callback.succeeded(); } } catch (IOException e) { // Too bad, can't close. } } + + /** + * A {@link BufferedWriter} that does not buffer. + * It is generally more efficient to buffer at the {@link Content.Sink} level, + * since frequently total content is smaller than a single buffer and + * the {@link Content.Sink} can turn a close into a last write that will avoid + * chunking the response if at all possible. However, {@link BufferedWriter} + * is in the API for {@link HttpResponse}, so we must return a writer of + * that type. + */ + private static class NonBufferedWriter extends BufferedWriter { + private final Writer writer; + + public NonBufferedWriter(Writer out) { + super(out, 1); + writer = out; + } + + @Override + public void write(int c) throws IOException { + writer.write(c); + } + + @Override + public void write(char[] cbuf) throws IOException { + writer.write(cbuf); + } + + @Override + public void write(char[] cbuf, int off, int len) throws IOException { + writer.write(cbuf, off, len); + } + + @Override + public void write(String str) throws IOException { + writer.write(str); + } + + @Override + public void write(String str, int off, int len) throws IOException { + writer.write(str, off, len); + } + + @Override + public Writer append(CharSequence csq) throws IOException { + return writer.append(csq); + } + + @Override + public Writer append(CharSequence csq, int start, int end) throws IOException { + return writer.append(csq, start, end); + } + + @Override + public Writer append(char c) throws IOException { + return writer.append(c); + } + + @Override + public void flush() throws IOException { + writer.flush(); + } + + @Override + public void close() throws IOException { + writer.close(); + } + + @Override + public void newLine() throws IOException { + writer.write(System.lineSeparator()); + } + } } From 1c142743c099c622d8d6d352aa3368bbbad4c7c2 Mon Sep 17 00:00:00 2001 From: gregw Date: Wed, 27 Sep 2023 08:45:32 +1000 Subject: [PATCH 5/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. Several optimizations are included, including the optimization that small writes can be buffered in such a way to avoid chunked responses when possible. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../invoker/http/HttpResponseImpl.java | 4 +- .../functions/invoker/IntegrationTest.java | 51 ++++++++++++++++++- .../invoker/testfunctions/BufferedWrites.java | 27 ++++++++++ 3 files changed, 78 insertions(+), 4 deletions(-) create mode 100644 invoker/core/src/test/java/com/google/cloud/functions/invoker/testfunctions/BufferedWrites.java diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index 81d8c5cf..fac96e19 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -116,9 +116,7 @@ public synchronized BufferedWriter getWriter() throws IOException { */ public void close(Callback callback) { try { - if (writer != null) { - writer.flush(); - } + // The writer has been constructed to do no buffering, so it does not need to be flushed if (contentSinkOutputStream != null) { // Do an asynchronous close, so large buffered content may be written without blocking contentSinkOutputStream.close(callback); diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java index a4b3300d..15cc87e8 100644 --- a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java @@ -167,6 +167,8 @@ abstract static class TestCase { abstract int expectedResponseCode(); + abstract Optional> expectedResponseHeaders(); + abstract Optional expectedResponseText(); abstract Optional expectedJson(); @@ -186,6 +188,7 @@ static Builder builder() { .setUrl("/") .setRequestText("") .setExpectedResponseCode(HttpStatus.OK_200) + .setExpectedResponseHeaders(ImmutableMap.of()) .setExpectedResponseText("") .setHttpContentType("text/plain") .setHttpHeaders(ImmutableMap.of()); @@ -204,6 +207,8 @@ Builder setRequestText(String text) { abstract Builder setExpectedResponseCode(int x); + abstract Builder setExpectedResponseHeaders(Map x); + abstract Builder setExpectedResponseText(String x); abstract Builder setExpectedResponseText(Optional x); @@ -249,11 +254,44 @@ public void helloWorld() throws Exception { testHttpFunction( fullTarget("HelloWorld"), ImmutableList.of( - TestCase.builder().setExpectedResponseText("hello\n").build(), + TestCase.builder() + .setExpectedResponseText("hello\n") + .setHttpHeaders(ImmutableMap.of( + "Content-Length", "*")) + .build(), FAVICON_TEST_CASE, ROBOTS_TXT_TEST_CASE)); } + @Test + public void bufferedWrites() throws Exception { + // This test checks that writes are buffered and are written + // in an efficient way with known content-length if possible. + testHttpFunction( + fullTarget("BufferedWrites"), + ImmutableList.of( + TestCase.builder() + .setUrl("/target?writes=2") + .setExpectedResponseText("write 0\nwrite 1\n") + .setExpectedResponseHeaders(ImmutableMap.of( + "x-write-0", "true", + "x-write-1", "true", + "x-written", "true", + "Content-Length", "16" + )) + .build(), + TestCase.builder() + .setUrl("/target?writes=2&flush=true") + .setExpectedResponseText("write 0\nwrite 1\n") + .setExpectedResponseHeaders(ImmutableMap.of( + "x-write-0", "true", + "x-write-1", "true", + "x-written", "-", + "Transfer-Encoding", "chunked")) + .build() + )); + } + @Test public void exceptionHttp() throws Exception { String exceptionExpectedOutput = @@ -696,6 +734,17 @@ private void testFunction( .withMessage("Response to %s is %s %s", uri, response.getStatus(), response.getReason()) .that(response.getStatus()) .isEqualTo(testCase.expectedResponseCode()); + testCase.expectedResponseHeaders().ifPresent(map -> { + for (Map.Entry entry : map.entrySet()) { + if ("*".equals(entry.getValue())) { + expect.that(response.getHeaders().getFieldNamesCollection()).contains(entry.getKey()); + } else if ("-".equals(entry.getValue())) { + expect.that(response.getHeaders().getFieldNamesCollection()).doesNotContain(entry.getKey()); + } else { + expect.that(response.getHeaders().getValuesList(entry.getKey())).contains(entry.getValue()); + } + } + }); testCase .expectedResponseText() .ifPresent(text -> expect.that(response.getContentAsString()).isEqualTo(text)); diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/testfunctions/BufferedWrites.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/testfunctions/BufferedWrites.java new file mode 100644 index 00000000..a7989a74 --- /dev/null +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/testfunctions/BufferedWrites.java @@ -0,0 +1,27 @@ +package com.google.cloud.functions.invoker.testfunctions; + +import com.google.cloud.functions.HttpFunction; +import com.google.cloud.functions.HttpRequest; +import com.google.cloud.functions.HttpResponse; +import java.io.BufferedWriter; +import java.util.List; +import java.util.Map; + +public class BufferedWrites implements HttpFunction { + @Override + public void service(HttpRequest request, HttpResponse response) throws Exception { + Map> queryParameters = request.getQueryParameters(); + int writes = Integer.parseInt(request.getFirstQueryParameter("writes").orElse("0")); + boolean flush = Boolean.parseBoolean(request.getFirstQueryParameter("flush").orElse("false")); + + BufferedWriter writer = response.getWriter(); + for (int i = 0; i < writes; i++) { + response.appendHeader("x-write-" + i, "true"); + writer.write("write " + i + "\n"); + } + if (flush) { + writer.flush(); + } + response.appendHeader("x-written", "true"); + } +} From 2e99c435d6df2071c2046f55cda11c6f225b7e23 Mon Sep 17 00:00:00 2001 From: gregw Date: Wed, 27 Sep 2023 09:07:25 +1000 Subject: [PATCH 6/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. Several optimizations are included, including the optimization that small writes can be buffered in such a way to avoid chunked responses when possible. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../com/google/cloud/functions/invoker/IntegrationTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java index 15cc87e8..a144d5d5 100644 --- a/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java +++ b/invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java @@ -255,9 +255,9 @@ public void helloWorld() throws Exception { fullTarget("HelloWorld"), ImmutableList.of( TestCase.builder() - .setExpectedResponseText("hello\n") - .setHttpHeaders(ImmutableMap.of( + .setExpectedResponseHeaders(ImmutableMap.of( "Content-Length", "*")) + .setExpectedResponseText("hello\n") .build(), FAVICON_TEST_CASE, ROBOTS_TXT_TEST_CASE)); From a2f9871c1bdac273394df78bbea280338dc4c788 Mon Sep 17 00:00:00 2001 From: gregw Date: Wed, 27 Sep 2023 12:04:45 +1000 Subject: [PATCH 7/7] chore(implementation)!: use Jetty-12 core without servlets Port the invoker to upgrade to Eclipse Jetty-12 version 12. Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided. Several optimizations are included, including the optimization that small writes can be buffered in such a way to avoid chunked responses when possible. BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12. --- .../functions/invoker/http/HttpResponseImpl.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java index fac96e19..60216514 100644 --- a/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java +++ b/invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java @@ -36,7 +36,7 @@ public class HttpResponseImpl implements HttpResponse { private final Response response; - private ContentSinkOutputStream contentSinkOutputStream; + private ContentSinkOutputStream outputStream; private BufferedWriter writer; private Charset charset; @@ -84,22 +84,22 @@ public Map> getHeaders() { public OutputStream getOutputStream() { if (writer != null) { throw new IllegalStateException("getWriter called"); - } else if (contentSinkOutputStream == null) { + } else if (outputStream == null) { Request request = response.getRequest(); int outputBufferSize = request.getConnectionMetaData().getHttpConfiguration() .getOutputBufferSize(); BufferedContentSink bufferedContentSink = new BufferedContentSink(response, request.getComponents().getByteBufferPool(), false, outputBufferSize / 2, outputBufferSize); - contentSinkOutputStream = new ContentSinkOutputStream(bufferedContentSink); + outputStream = new ContentSinkOutputStream(bufferedContentSink); } - return contentSinkOutputStream; + return outputStream; } @Override public synchronized BufferedWriter getWriter() throws IOException { if (writer == null) { - if (contentSinkOutputStream != null) { + if (outputStream != null) { throw new IllegalStateException("getOutputStream called"); } @@ -117,9 +117,9 @@ public synchronized BufferedWriter getWriter() throws IOException { public void close(Callback callback) { try { // The writer has been constructed to do no buffering, so it does not need to be flushed - if (contentSinkOutputStream != null) { + if (outputStream != null) { // Do an asynchronous close, so large buffered content may be written without blocking - contentSinkOutputStream.close(callback); + outputStream.close(callback); } else { callback.succeeded(); }