Support reading HTTP request trailers on the server

HttpServerRequest had no way to access trailers, while HttpClientResponse
exposes trailers() and getTrailer(). Trailers were parsed by Netty and then
dropped: Http1ServerConnection carried a "//TODO chunk trailers" comment, and
the HTTP/2 codec connection noted "not implemented yet (in api)".

The public interface gains getTrailer(String) and trailers(), mirroring
HttpClientResponse, and returning an empty map before the request has ended
rather than null.

For HTTP/1 the trailers travel through the request's inbound message queue,
as the client already does, so they cannot be observed before the preceding
data has been delivered. For HTTP/2 the codec connection now forwards the
trailing HEADERS frame to the stream instead of signalling empty trailers;
the multiplex, HTTP/3 and QUIC paths already forwarded them, and
HttpServerRequestImpl already received the MultiMap and discarded it.

Fixes #5253

Author: jnbdz

vertx-core/src/main/java/io/vertx/core/http/HttpServerRequest.java

@@ -149,6 +149,29 @@ default boolean isSSL() {
   @CacheReturn
   HttpServerResponse response();
 
+  /**
+   * Return the first trailer value with the specified name
+   * <p>
+   * Trailers are only available after the request has been fully received, i.e. after the
+   * {@link #endHandler(Handler) end handler} has been called.
+   *
+   * @param trailerName  the trailer name
+   * @return the trailer value
+   */
+  @Nullable String getTrailer(String trailerName);
+
+  /**
+   * Return the trailers.
+   * <p>
+   * Trailers are only available after the request has been fully received, i.e. after the
+   * {@link #endHandler(Handler) end handler} has been called. Before that, and for requests
+   * that carry no trailers, the returned map is empty.
+   *
+   * @return the trailers
+   */
+  @CacheReturn
+  MultiMap trailers();
+
   /**
    * Override the charset to use for decoding the query parameter map, when none is set, {@code UTF8} is used.
    *

vertx-core/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java

@@ -27,6 +27,7 @@
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpVersion;
 import io.vertx.core.http.*;
+import io.vertx.core.http.impl.headers.HeadersAdaptor;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.http.QueryParamDecoder;
 import io.vertx.core.net.HostAndPort;
@@ -60,6 +61,7 @@ public class HttpServerRequestImpl extends HttpServerRequestBase {
   private HostAndPort realAuthority;
   private String absoluteURI;
   private MultiMap attributes;
+  private MultiMap trailers;
   private HttpEventHandler eventHandler;
   private boolean ended;
   private Handler<HttpServerFileUpload> uploadHandler;
@@ -188,6 +190,13 @@ public void handleData(Buffer data) {
   public void handleTrailers(MultiMap trailers) {
     HttpEventHandler handler;
     synchronized (connection) {
+      // Setting the trailers must not race with trailers(), where the field can escape:
+      // if the user already obtained the empty map, update it in place instead.
+      if (this.trailers == null) {
+        this.trailers = trailers;
+      } else if (this.trailers != trailers) {
+        this.trailers.setAll(trailers);
+      }
       ended = true;
       if (postRequestDecoder != null) {
         try {
@@ -374,6 +383,25 @@ public MultiMap headers() {
     return headersMap;
   }
 
+  @Override
+  public MultiMap trailers() {
+    synchronized (connection) {
+      if (trailers == null) {
+        trailers = new HeadersAdaptor(new DefaultHttpHeaders());
+      }
+      return trailers;
+    }
+  }
+
+  @Override
+  public String getTrailer(String trailerName) {
+    MultiMap trailers;
+    synchronized (connection) {
+      trailers = this.trailers;
+    }
+    return trailers != null ? trailers.get(trailerName) : null;
+  }
+
   @Override
   public SocketAddress remoteAddress() {
     return super.remoteAddress();

vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerConnection.java

@@ -246,18 +246,21 @@ private void onContent(Object msg) {
     Buffer buffer = BufferInternal.safeBuffer(content.content());
     Http1ServerRequest request = requestInProgress;
     request.handleContent(buffer);
-    //TODO chunk trailers
     if (content instanceof LastHttpContent) {
-      onEnd();
+      onEnd(((LastHttpContent) content).trailingHeaders());
     }
   }
 
   private void onEnd() {
+    onEnd(null);
+  }
+
+  private void onEnd(HttpHeaders trailers) {
     boolean tryClose;
     Http1ServerRequest request = requestInProgress;
     requestInProgress = null;
     tryClose = (wantClose || shutdownInitiated != null) && responseInProgress == null;
-    request.handleEnd();
+    request.handleEnd(trailers);
     if (tryClose) {
       closeInternal();
     }

vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerRequest.java

@@ -71,6 +71,7 @@ public class Http1ServerRequest extends HttpServerRequestBase implements io.vert
 
   // Cache this for performance
   private MultiMap headers;
+  private MultiMap trailers;
   private String absoluteURI;
 
   private HttpEventHandler eventHandler;
@@ -104,6 +105,11 @@ private InboundMessageQueue<Object> queue(boolean create) {
             protected void handleMessage(Object elt) {
               if (elt == InboundBuffer.END_SENTINEL) {
                 onEnd();
+              } else if (elt instanceof MultiMap) {
+                // Trailers travel through the queue so they cannot be observed before the
+                // preceding data has been delivered.
+                setTrailers((MultiMap) elt);
+                onEnd();
               } else {
                 onData((Buffer) elt);
               }
@@ -155,22 +161,40 @@ void handleContent(Buffer buffer) {
     }
   }
 
-  void handleEnd() {
+  void handleEnd(HttpHeaders nettyTrailers) {
+    Object end = nettyTrailers != null && !nettyTrailers.isEmpty()
+      ? new HeadersAdaptor(nettyTrailers)
+      : InboundBuffer.END_SENTINEL;
     InboundMessageQueue<Object> queue = queue(false);
     if (queue != null) {
-      handleEnd(queue);
+      handleEnd(queue, end);
     } else {
+      if (end != InboundBuffer.END_SENTINEL) {
+        setTrailers((MultiMap) end);
+      }
       context.execute(this, Http1ServerRequest::onEnd);
     }
   }
 
-  private void handleEnd(InboundMessageQueue<Object> queue) {
-    boolean drain = queue.add(InboundBuffer.END_SENTINEL);
+  private void handleEnd(InboundMessageQueue<Object> queue, Object end) {
+    boolean drain = queue.add(end);
     if (drain) {
       queue.drain();
     }
   }
 
+  private void setTrailers(MultiMap trailers) {
+    synchronized (conn) {
+      // Must not race with trailers(), where the field can escape: if the user already
+      // obtained the empty map, update it in place instead of replacing it.
+      if (this.trailers == null) {
+        this.trailers = trailers;
+      } else if (this.trailers != trailers) {
+        this.trailers.setAll(trailers);
+      }
+    }
+  }
+
   private void check100() {
     if (HttpUtil.is100ContinueExpected(request)) {
       response.writeContinue();
@@ -282,6 +306,25 @@ public HostAndPort authority(boolean real) {
     return real ? null : authority();
   }
 
+  @Override
+  public MultiMap trailers() {
+    synchronized (conn) {
+      if (trailers == null) {
+        trailers = new HeadersAdaptor(new DefaultHttpHeaders());
+      }
+      return trailers;
+    }
+  }
+
+  @Override
+  public String getTrailer(String trailerName) {
+    MultiMap trailers;
+    synchronized (conn) {
+      trailers = this.trailers;
+    }
+    return trailers != null ? trailers.get(trailerName) : null;
+  }
+
   @Override
   public long bytesRead() {
     synchronized (conn) {

vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ServerConnectionImpl.java

@@ -148,8 +148,10 @@ protected synchronized void onHeadersRead(int streamId, Http2Headers headers, St
       stream.context().execute(stream.unwrap(), streamHandler);
       stream.onHeaders(headersMap);
     } else {
-      // Http server request trailer - not implemented yet (in api)
+      // Trailing HEADERS frame: carry the trailers to the request instead of discarding them.
       stream = nettyStream.getProperty(streamKey);
+      stream.onTrailers(new HttpHeaders(headers));
+      return;
     }
     if (endOfStream) {
       stream.onTrailers();

vertx-core/src/main/java/io/vertx/core/internal/http/HttpServerRequestWrapper.java

@@ -130,6 +130,16 @@ public MultiMap headers() {
     return delegate.headers();
   }
 
+  @Override
+  public String getTrailer(String trailerName) {
+    return delegate.getTrailer(trailerName);
+  }
+
+  @Override
+  public MultiMap trailers() {
+    return delegate.trailers();
+  }
+
   @Override
   public HttpServerRequest setParamsCharset(String charset) {
     return delegate.setParamsCharset(charset);

vertx-core/src/test/java/io/vertx/tests/http/Http1xTest.java

@@ -3395,6 +3395,50 @@ public void testTooLongContentInHttpServerRequest(Checkpoint checkpoint) throws
       .write("POST / HTTP/1.1\r\nContent-Length: 4\r\n\r\ntoolong\r\n")
       .await();
   }
+  @Test
+  public void testRequestTrailers(Checkpoint checkpoint) throws Exception {
+    server.requestHandler(req -> {
+      req.endHandler(v -> {
+        assertEquals("chunky", req.getTrailer("X-Trailer"));
+        assertEquals("other", req.getTrailer("X-Other"));
+        assertEquals(2, req.trailers().size());
+        checkpoint.succeed();
+      });
+      req.response().end();
+    });
+    startServer(testAddress);
+    NetClient client = vertx.createNetClient();
+    NetSocket so = client.connect(testAddress).await();
+    so.write("POST / HTTP/1.1\r\n" +
+      "Host: localhost\r\n" +
+      "Transfer-Encoding: chunked\r\n" +
+      "Trailer: X-Trailer\r\n" +
+      "\r\n" +
+      "5\r\nhello\r\n" +
+      "0\r\n" +
+      "X-Trailer: chunky\r\n" +
+      "X-Other: other\r\n" +
+      "\r\n");
+    checkpoint.awaitSuccess();
+  }
+
+  @Test
+  public void testRequestNoTrailers(Checkpoint checkpoint) throws Exception {
+    server.requestHandler(req -> {
+      req.endHandler(v -> {
+        assertTrue(req.trailers().isEmpty());
+        assertNull(req.getTrailer("X-Trailer"));
+        checkpoint.succeed();
+      });
+      req.response().end();
+    });
+    startServer(testAddress);
+    client.request(requestOptions)
+      .compose(req -> req.send(Buffer.buffer("hello")))
+      .await();
+    checkpoint.awaitSuccess();
+  }
+
   @Test
   public void testInvalidTrailerInHttpServerRequest(Checkpoint checkpoint) throws Exception {
     testHttpServerRequestDecodeError(checkpoint, so -> {