Use ResponseLifecycle from core http-server

Instead of servlet-specific body encoding logic, use the ResponseLifecycle introduced in core by micronaut-projects/micronaut-core#11342 (pending review). Then we only have to take care of writing the ByteBody to the servlet response.

Some other related changes:

- Improved async reading/writing support, deprecate StreamedServletMessage
- Rework HttpResponse lifecycle management: Core mostly expects HttpResponseFactory to return new, independent responses, but servlet returned views of the same ServletHttpResponse each time (with shared headers and such). New approach is to have only the latest HttpResponse backed by the real ServletHttpResponse, any previous responses are copied snapshots
- Deprecate and stop using ServletResponseEncoder, it is replaced by core ResponseBodyEncoder.

These changes fix all the remaining TCK failures that relate to response handling, except for FilterProxyTest.

http-poja-apache/src/main/java/io/micronaut/http/poja/apache/ApacheResponseContext.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2017-2024 original 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
+ *
+ * https://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.micronaut.http.poja.apache;
+
+import io.micronaut.http.HttpHeaders;
+import io.micronaut.http.server.exceptions.HttpServerException;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.impl.io.ChunkedOutputStream;
+import org.apache.hc.core5.http.impl.io.ContentLengthOutputStream;
+import org.apache.hc.core5.http.impl.io.DefaultHttpResponseWriter;
+import org.apache.hc.core5.http.impl.io.SessionOutputBufferImpl;
+import org.apache.hc.core5.http.io.SessionOutputBuffer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.OutputStream;
+
+final class ApacheResponseContext implements Closeable {
+    boolean connectionClose = false;
+    ApacheServletHttpResponse<?> primaryResponse;
+
+    private final SessionOutputBuffer outputBuffer;
+    private final OutputStream out;
+    private OutputStream bodyStream;
+
+    ApacheResponseContext(ApacheServletConfiguration configuration, OutputStream out) {
+        this.out = out;
+        outputBuffer = new SessionOutputBufferImpl(configuration.outputBufferSize());
+    }
+
+    boolean isCommitted() {
+        return bodyStream != null;
+    }
+
+    OutputStream commit(ClassicHttpResponse headers) throws IOException {
+        if (isCommitted()) {
+            throw new IllegalStateException("Response has already been committed");
+        }
+
+        headers.removeHeaders(HttpHeaders.TRANSFER_ENCODING);
+        Header contentLengthStr = headers.getFirstHeader(HttpHeaders.CONTENT_LENGTH);
+        long contentLength = contentLengthStr == null ? -1 : Long.parseLong(contentLengthStr.getValue());
+        if (contentLength < 0) {
+            headers.removeHeaders(HttpHeaders.CONTENT_LENGTH);
+            headers.addHeader(HttpHeaders.TRANSFER_ENCODING, "chunked");
+        }
+
+        Header connection = headers.getFirstHeader(HttpHeaders.CONNECTION);
+        if (connection != null && connection.getValue().equalsIgnoreCase("close")) {
+            connectionClose = true;
+        }
+
+        DefaultHttpResponseWriter responseWriter = new DefaultHttpResponseWriter();
+        try {
+            responseWriter.write(headers, outputBuffer, out);
+        } catch (HttpException e) {
+            throw new HttpServerException("Could not write response headers", e);
+        }
+
+        OutputStream s = contentLength < 0 ? new ChunkedOutputStream(outputBuffer, out, 0) : new ContentLengthOutputStream(outputBuffer, out, contentLength);
+        bodyStream = s;
+        return s;
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (bodyStream != null) {
+            bodyStream.close();
+        }
+        outputBuffer.flush(out);
+
+        if (connectionClose) {
+            out.close();
+        }
+    }
+}

http-poja-apache/src/main/java/io/micronaut/http/poja/apache/ApacheServerlessApplication.java

@@ -17,31 +17,24 @@
 
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.core.convert.ConversionService;
+import io.micronaut.core.io.buffer.ByteArrayBufferFactory;
 import io.micronaut.core.io.buffer.ByteBufferFactory;
 import io.micronaut.http.HttpStatus;
-import io.micronaut.http.MediaType;
 import io.micronaut.http.codec.MediaTypeCodecRegistry;
 import io.micronaut.http.poja.PojaHttpServerlessApplication;
-import io.micronaut.http.poja.apache.exception.ApacheServletBadRequestException;
-import io.micronaut.http.server.exceptions.HttpServerException;
 import io.micronaut.inject.qualifiers.Qualifiers;
 import io.micronaut.runtime.ApplicationConfiguration;
 import io.micronaut.scheduling.TaskExecutors;
-import io.micronaut.servlet.http.ByteArrayBufferFactory;
 import io.micronaut.servlet.http.ServletHttpHandler;
 import jakarta.inject.Singleton;
-import org.apache.hc.core5.http.ClassicHttpResponse;
-import org.apache.hc.core5.http.HttpEntity;
-import org.apache.hc.core5.http.HttpException;
-import org.apache.hc.core5.http.impl.io.DefaultHttpResponseWriter;
 import org.apache.hc.core5.http.impl.io.SessionInputBufferImpl;
-import org.apache.hc.core5.http.impl.io.SessionOutputBufferImpl;
 import org.apache.hc.core5.http.io.SessionInputBuffer;
-import org.apache.hc.core5.http.io.SessionOutputBuffer;
+import org.apache.hc.core5.http.message.BasicClassicHttpResponse;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ExecutorService;
 
 /**
@@ -78,46 +71,34 @@ public ApacheServerlessApplication(ApplicationContext applicationContext,
     }
 
     @Override
-    protected void handleSingleRequest(
+    protected boolean handleSingleRequest(
             ServletHttpHandler<ApacheServletHttpRequest<?>, ApacheServletHttpResponse<?>> servletHttpHandler,
             InputStream in,
             OutputStream out
     ) throws IOException {
-        ApacheServletHttpResponse<?> response = new ApacheServletHttpResponse<>(conversionService);
-        try {
-            // The buffer is initialized only once
-            if (sessionInputBuffer == null) {
-                sessionInputBuffer = new SessionInputBufferImpl(configuration.inputBufferSize());
+        try (ApacheResponseContext responseContext = new ApacheResponseContext(configuration, out)) {
+            try {
+                // The buffer is initialized only once
+                if (sessionInputBuffer == null) {
+                    sessionInputBuffer = new SessionInputBufferImpl(configuration.inputBufferSize());
+                }
+                ApacheServletHttpRequest exchange = new ApacheServletHttpRequest<>(
+                    in, responseContext, sessionInputBuffer, conversionService, codecRegistry, ioExecutor, byteBufferFactory
+                );
+                servletHttpHandler.service(exchange);
+                if (!responseContext.isCommitted()) {
+                    responseContext.primaryResponse.getOutputStream(); // this causes the commit
+                }
+            } catch (Exception e) {
+                if (!responseContext.isCommitted()) {
+                    try (OutputStream os = responseContext.commit(new BasicClassicHttpResponse(HttpStatus.BAD_REQUEST.getCode()))) {
+                        os.write(e.getMessage().getBytes(StandardCharsets.UTF_8));
+                    }
+                }
+                throw e;
             }
-            ApacheServletHttpRequest exchange = new ApacheServletHttpRequest<>(
-                in, sessionInputBuffer, conversionService, codecRegistry, ioExecutor, byteBufferFactory, response, configuration
-            );
-            servletHttpHandler.service(exchange);
-        } catch (ApacheServletBadRequestException e) {
-            response.status(HttpStatus.BAD_REQUEST);
-            response.contentType(MediaType.TEXT_PLAIN_TYPE);
-            response.getOutputStream().write(e.getMessage().getBytes());
-            writeResponse(response.getNativeResponse(), out);
-            throw e;
+            return !responseContext.connectionClose;
         }
-        writeResponse(response.getNativeResponse(), out);
-    }
-
-    private void writeResponse(ClassicHttpResponse response, OutputStream out) throws IOException {
-        SessionOutputBuffer buffer = new SessionOutputBufferImpl(configuration.outputBufferSize());
-        DefaultHttpResponseWriter responseWriter = new DefaultHttpResponseWriter();
-        try {
-            responseWriter.write(response, buffer, out);
-        } catch (HttpException e) {
-            throw new HttpServerException("Could not write response body", e);
-        }
-        buffer.flush(out);
-
-        HttpEntity entity = response.getEntity();
-        if (entity != null) {
-            entity.writeTo(out);
-        }
-        out.flush();
     }
 
     @Override

http-poja-apache/src/main/java/io/micronaut/http/poja/apache/ApacheServletHttpRequest.java

@@ -25,6 +25,7 @@
 import io.micronaut.http.MutableHttpParameters;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.body.ByteBody;
+import io.micronaut.http.body.ByteBodyFactory;
 import io.micronaut.http.body.stream.InputStreamByteBody;
 import io.micronaut.http.codec.MediaTypeCodecRegistry;
 import io.micronaut.http.cookie.Cookie;
@@ -35,6 +36,7 @@
 import io.micronaut.http.poja.util.MultiValueHeaders;
 import io.micronaut.http.poja.util.MultiValuesQueryParameters;
 import io.micronaut.http.simple.cookies.SimpleCookies;
+import io.micronaut.servlet.http.ServletHttpResponse;
 import org.apache.hc.core5.http.ClassicHttpRequest;
 import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.Header;
@@ -73,6 +75,7 @@ public final class ApacheServletHttpRequest<B> extends PojaHttpRequest<B, Classi
     private static final String TRANSFER_ENCODING_CHUNKED = "chunked";
 
     private final ClassicHttpRequest request;
+    private final ApacheResponseContext responseContext;
 
     private final HttpMethod method;
     private URI uri;
@@ -82,28 +85,30 @@ public final class ApacheServletHttpRequest<B> extends PojaHttpRequest<B, Classi
 
     private final ByteBody byteBody;
 
+    private ApacheServletHttpResponse<?> primaryResponse;
+
     /**
      * Create an Apache-based request.
      *
      * @param inputStream The input stream
+     * @param responseContext The response context
+     * @param sessionInputBuffer Input buffer for parsing
      * @param conversionService The conversion service
      * @param codecRegistry The media codec registry
      * @param ioExecutor The executor service
      * @param byteBufferFactory The byte buffer factory
-     * @param response The response
-     * @param configuration The configuration
      */
     public ApacheServletHttpRequest(
         InputStream inputStream,
+        ApacheResponseContext responseContext,
         SessionInputBuffer sessionInputBuffer,
         ConversionService conversionService,
         MediaTypeCodecRegistry codecRegistry,
         ExecutorService ioExecutor,
-        ByteBufferFactory<?, ?> byteBufferFactory,
-        ApacheServletHttpResponse<?> response,
-        ApacheServletConfiguration configuration
+        ByteBufferFactory<?, ?> byteBufferFactory
     ) {
-        super(conversionService, codecRegistry, response);
+        super(conversionService, codecRegistry);
+        this.responseContext = responseContext;
         DefaultHttpRequestParser parser = new DefaultHttpRequestParser();
 
         try {
@@ -125,12 +130,20 @@ public ApacheServletHttpRequest(
         queryParameters = parseQueryParameters(uri, conversionService);
         cookies = parseCookies(request, conversionService);
 
+        Header connection = request.getFirstHeader(HttpHeaders.CONNECTION);
+        if (connection != null && connection.getValue().equalsIgnoreCase("close")) {
+            responseContext.connectionClose = true;
+        }
+
         long contentLength = getContentLength();
+        if (!getMethod().permitsRequestBody()) {
+            contentLength = 0;
+        }
         OptionalLong optionalContentLength = contentLength >= 0 ? OptionalLong.of(contentLength) : OptionalLong.empty();
         InputStream bodyStream = createBodyStream(inputStream, contentLength, sessionInputBuffer);
         byteBody = InputStreamByteBody.create(
-            bodyStream, optionalContentLength, ioExecutor, byteBufferFactory
-        );
+            bodyStream, optionalContentLength, ioExecutor, ByteBodyFactory.createDefault(byteBufferFactory));
+        primaryResponse = new ApacheServletHttpResponse<>(responseContext, conversionService);
     }
 
     /**
@@ -156,6 +169,18 @@ private InputStream createBodyStream(InputStream inputStream, long contentLength
         return bodyStream;
     }
 
+    @Override
+    public ServletHttpResponse<ClassicHttpResponse, ?> getResponse() {
+        return primaryResponse;
+    }
+
+    @Override
+    public ServletHttpResponse<ClassicHttpResponse, ?> createResponse() {
+        ApacheServletHttpResponse<Object> r = new ApacheServletHttpResponse<>(responseContext, conversionService);
+        primaryResponse = r;
+        return r;
+    }
+
     @Override
     public ClassicHttpRequest getNativeRequest() {
         return request;

http-poja-apache/src/main/java/io/micronaut/http/poja/apache/ApacheServletHttpResponse.java

@@ -21,23 +21,20 @@
 import io.micronaut.core.convert.ConversionService;
 import io.micronaut.core.convert.value.MutableConvertibleValues;
 import io.micronaut.core.convert.value.MutableConvertibleValuesMap;
-import io.micronaut.http.HttpHeaders;
 import io.micronaut.http.HttpStatus;
 import io.micronaut.http.MutableHttpHeaders;
 import io.micronaut.http.MutableHttpResponse;
 import io.micronaut.http.cookie.Cookie;
 import io.micronaut.http.poja.PojaHttpResponse;
 import io.micronaut.http.simple.SimpleHttpHeaders;
 import org.apache.hc.core5.http.ClassicHttpResponse;
-import org.apache.hc.core5.http.ContentType;
-import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
 import org.apache.hc.core5.http.message.BasicClassicHttpResponse;
 
 import java.io.BufferedWriter;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.io.PrintWriter;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
 import java.util.Optional;
 
 /**
@@ -48,11 +45,12 @@
  * @since 4.10.0
  */
 @Internal
-public final class ApacheServletHttpResponse<T> extends PojaHttpResponse<T, ClassicHttpResponse> {
+final class ApacheServletHttpResponse<T> extends PojaHttpResponse<T, ClassicHttpResponse> {
+
+    private final ApacheResponseContext responseContext;
 
     private int code = HttpStatus.OK.getCode();
     private String reasonPhrase = HttpStatus.OK.getReason();
-    private final ByteArrayOutputStream out = new ByteArrayOutputStream();
 
     private final SimpleHttpHeaders headers;
     private final MutableConvertibleValues<Object> attributes = new MutableConvertibleValuesMap<>();
@@ -63,35 +61,27 @@ public final class ApacheServletHttpResponse<T> extends PojaHttpResponse<T, Clas
      *
      * @param conversionService The conversion service
      */
-    public ApacheServletHttpResponse(ConversionService conversionService) {
+    ApacheServletHttpResponse(ApacheResponseContext responseContext, ConversionService conversionService) {
+        this.responseContext = responseContext;
         this.headers = new SimpleHttpHeaders(conversionService);
+        responseContext.primaryResponse = this;
     }
 
     @Override
     public ClassicHttpResponse getNativeResponse() {
-        headers.remove(HttpHeaders.CONTENT_LENGTH);
-        headers.add(HttpHeaders.CONTENT_LENGTH, String.valueOf(out.size()));
-        if ("chunked".equalsIgnoreCase(headers.get(HttpHeaders.TRANSFER_ENCODING))) {
-            headers.remove(HttpHeaders.TRANSFER_ENCODING);
-        }
-
         BasicClassicHttpResponse response = new BasicClassicHttpResponse(code, reasonPhrase);
         headers.forEachValue(response::addHeader);
-        ContentType contentType = headers.getContentType().map(ContentType::parse)
-            .orElse(ContentType.APPLICATION_JSON);
-        ByteArrayEntity body = new ByteArrayEntity(out.toByteArray(), contentType);
-        response.setEntity(body);
         return response;
     }
 
     @Override
     public OutputStream getOutputStream() throws IOException {
-        return out;
+        return responseContext.commit(getNativeResponse());
     }
 
     @Override
     public BufferedWriter getWriter() throws IOException {
-        return new BufferedWriter(new PrintWriter(out));
+        return new BufferedWriter(new OutputStreamWriter(getOutputStream(), StandardCharsets.UTF_8));
     }
 
     @Override

http-poja-apache/src/test/groovy/io/micronaut/http/poja/SimpleServerFailedParseSpec.groovy

@@ -23,10 +23,13 @@ class SimpleServerFailedParseSpec extends Specification {
         then:
         response == SimpleServerSpec.unindent("""
         HTTP/1.1 400 Bad Request\r
-        Content-Length: 32\r
-        Content-Type: text/plain\r
+        Transfer-Encoding: chunked\r
         \r
-        HTTP request could not be parsed""")
+        20\r
+        HTTP request could not be parsed\r
+        0\r
+        \r
+        """)
     }
 
 }