Support HTTP request trailers

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

@@ -234,6 +234,46 @@ public interface HttpClientRequest extends WriteStream<Buffer> {
   @Fluent
   HttpClientRequest putHeader(CharSequence name, Iterable<CharSequence> values);
 
+  /**
+   * @return The HTTP trailers
+   */
+  @CacheReturn
+  MultiMap trailers();
+
+  /**
+   * Put an HTTP trailer
+   * <p>
+   * Trailers are sent after the request body, so the request is always sent chunked when any
+   * trailer is set.
+   *
+   * @param name  the trailer name
+   * @param value  the trailer value
+   * @return a reference to this, so the API can be used fluently
+   */
+  @Fluent
+  HttpClientRequest putTrailer(String name, String value);
+
+  /**
+   * Like {@link #putTrailer(String, String)} but using CharSequence
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  @Fluent
+  HttpClientRequest putTrailer(CharSequence name, CharSequence value);
+
+  /**
+   * Like {@link #putTrailer(String, String)} but providing multiple values via a String Iterable
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  @Fluent
+  HttpClientRequest putTrailer(String name, Iterable<String> values);
+
+  /**
+   * Like {@link #putTrailer(String, Iterable)} but with CharSequence Iterable
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  @Fluent
+  HttpClientRequest putTrailer(CharSequence name, Iterable<CharSequence> value);
+
   /**
    * Set the trace operation of this request.
    *

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/HttpClientConnection.java

@@ -36,6 +36,11 @@ public interface HttpClientConnection extends HttpConnection {
 
   MultiMap newHttpRequestHeaders();
 
+  /**
+   * @return an empty trailer map suitable for this connection's protocol
+   */
+  MultiMap newHttpTrailers();
+
   /**
    * @return the number of active request/response (streams)
    */

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

@@ -56,6 +56,7 @@ public class HttpClientRequestImpl extends HttpClientRequestBase implements Http
   private int maxRedirects;
   private int numberOfRedirections;
   private final MultiMap headers;
+  private MultiMap trailers;
   private boolean trailersSent;
   private boolean headersSent;
   private StreamPriority priority;
@@ -205,6 +206,42 @@ public MultiMap headers() {
     return headers;
   }
 
+  @Override
+  public synchronized MultiMap trailers() {
+    if (trailers == null) {
+      trailers = stream.connection().newHttpTrailers();
+    }
+    return trailers;
+  }
+
+  @Override
+  public synchronized HttpClientRequest putTrailer(String name, String value) {
+    checkEnded();
+    trailers().set(name, value);
+    return this;
+  }
+
+  @Override
+  public synchronized HttpClientRequest putTrailer(CharSequence name, CharSequence value) {
+    checkEnded();
+    trailers().set(name, value);
+    return this;
+  }
+
+  @Override
+  public synchronized HttpClientRequest putTrailer(String name, Iterable<String> values) {
+    checkEnded();
+    trailers().set(name, values);
+    return this;
+  }
+
+  @Override
+  public synchronized HttpClientRequest putTrailer(CharSequence name, Iterable<CharSequence> value) {
+    checkEnded();
+    trailers().set(name, value);
+    return this;
+  }
+
   @Override
   public synchronized HttpClientRequest putHeader(String name, String value) {
     checkEnded();
@@ -531,6 +568,7 @@ private Future<Void> doWrite(Buffer buff, boolean end, boolean connect) {
     boolean writeHead;
     boolean writeEnd;
     boolean chunked;
+    MultiMap trailersToSend;
     synchronized (this) {
       if (reset != null) {
         return context.failedFuture(reset);
@@ -557,16 +595,31 @@ private Future<Void> doWrite(Buffer buff, boolean end, boolean connect) {
           }
         }
       }
+      // Trailers ride on the terminating chunk, so the request must be chunked: Netty
+      // silently drops trailing headers when a Content-Length framing is used.
+      // Captured under the lock, like every other value used after it, so the reference is
+      // safely published to whichever thread performs the write.
+      trailersToSend = end && !connect && !isConnect && trailers != null && !trailers.isEmpty()
+        ? trailers
+        : null;
       if (!headersSent) {
         if (!connect) {
-          boolean requiresContentLength = !this.chunked && !headers.contains(CONTENT_LENGTH);
-          if (end) {
-            if (buff != null && requiresContentLength) {
-              headers().set(CONTENT_LENGTH, HttpUtils.positiveLongToString(buff.length()));
+          if (trailersToSend != null) {
+            if (!this.chunked) {
+              headers.remove(CONTENT_LENGTH);
+              headers.set(TRANSFER_ENCODING, CHUNKED);
+              this.chunked = true;
+            }
+          } else {
+            boolean requiresContentLength = !this.chunked && !headers.contains(CONTENT_LENGTH);
+            if (end) {
+              if (buff != null && requiresContentLength) {
+                headers().set(CONTENT_LENGTH, HttpUtils.positiveLongToString(buff.length()));
+              }
+            } else if (requiresContentLength) {
+              headers.set(TRANSFER_ENCODING, CHUNKED);
+              this.chunked = true;
             }
-          } else if (requiresContentLength) {
-            headers.set(TRANSFER_ENCODING, CHUNKED);
-            this.chunked = true;
           }
         }
         headersSent = true;
@@ -576,7 +629,8 @@ private Future<Void> doWrite(Buffer buff, boolean end, boolean connect) {
         writeHead = false;
       }
       chunked = this.chunked;
-      writeEnd = !isConnect && end;
+      // Suppress the body's end marker so the trailers can carry it instead.
+      writeEnd = !isConnect && end && trailersToSend == null;
       trailersSent = end;
     }
 
@@ -595,6 +649,12 @@ private Future<Void> doWrite(Buffer buff, boolean end, boolean connect) {
       }
       future = stream.writeChunk(buff, writeEnd);
     }
+    if (trailersToSend != null) {
+      MultiMap t = trailersToSend;
+      // Composed rather than fired-and-forgotten: the request is only complete once the
+      // trailers are written, and a failed body write must not emit them.
+      future = future.compose(v -> stream.writeHeaders(t, true));
+    }
     if (end) {
       tryComplete();
     }

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

@@ -147,6 +147,31 @@ public HttpClientRequest putHeader(CharSequence name, Iterable<CharSequence> val
     throw new IllegalStateException();
   }
 
+  @Override
+  public MultiMap trailers() {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putTrailer(String name, String value) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putTrailer(CharSequence name, CharSequence value) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putTrailer(String name, Iterable<String> values) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putTrailer(CharSequence name, Iterable<CharSequence> value) {
+    throw new IllegalStateException();
+  }
+
   @Override
   public HttpClientRequest traceOperation(String op) {
     throw new IllegalStateException();

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

@@ -28,6 +28,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;
@@ -61,6 +62,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;
@@ -189,6 +191,11 @@ public void handleData(Buffer data) {
   public void handleTrailers(MultiMap trailers) {
     HttpEventHandler handler;
     synchronized (connection) {
+      if (this.trailers == null) {
+        this.trailers = trailers;
+      } else if (this.trailers != trailers) {
+        this.trailers.setAll(trailers);
+      }
       ended = true;
       if (postRequestDecoder != null) {
         try {
@@ -380,6 +387,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/HttpServerStream.java

@@ -34,7 +34,6 @@ public interface HttpServerStream extends HttpStream {
   HttpServerConnection connection();
 
   Future<Void> writeHead(HttpResponseHead head, Buffer chunk, boolean end);
-  Future<Void> writeHeaders(MultiMap headers, boolean end);
 
   Future<HttpServerStream> sendPush(HostAndPort authority, HttpMethod method, MultiMap headers, String path, StreamPriority priority);
 

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

@@ -44,6 +44,7 @@ public interface HttpStream {
   ByteBufAllocator allocator();
 
   Future<Void> writeChunk(Buffer buf, boolean end);
+  Future<Void> writeHeaders(MultiMap headers, boolean end);
   Future<Void> writeFrame(int type, int flags, Buffer payload);
   Future<Void> writeReset(long code);
 

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

@@ -82,6 +82,14 @@ public Future<Void> writeHead(HttpRequestHead request, boolean chunked, Buffer b
     return delegate.writeHead(request, chunked, buf, end, priority, connect);
   }
 
+  @Override
+  public Future<Void> writeHeaders(MultiMap headers, boolean end) {
+    if (end) {
+      endpointRequest.reportRequestEnd();
+    }
+    return delegate.writeHeaders(headers, end);
+  }
+
   @Override
   public Future<Void> writeChunk(Buffer buf, boolean end) {
     if (end) {

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

@@ -32,6 +32,7 @@
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.vertx.core.*;
+import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.*;
 import io.vertx.core.http.HttpMethod;
@@ -145,6 +146,11 @@ public MultiMap newHttpRequestHeaders() {
     return Http1xHeaders.httpHeaders();
   }
 
+  @Override
+  public MultiMap newHttpTrailers() {
+    return Http1xHeaders.httpHeaders();
+  }
+
   @Override
   public HostAndPort authority() {
     return authority;
@@ -396,6 +402,26 @@ public void cancel(Throwable cause) {
     });
   }
 
+  private void writeTrailers(Stream stream, io.netty.handler.codec.http.HttpHeaders trailers, Promise<Void> listener) {
+    writeToChannel(new MessageWrite() {
+      @Override
+      public void write() {
+        if (stream.reset) {
+          listener.fail("Stream reset");
+          return;
+        }
+        assert current == stream;
+        unsafeWrite(new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER, trailers), listener);
+        endRequest(stream);
+      }
+
+      @Override
+      public void cancel(Throwable cause) {
+        listener.fail(cause);
+      }
+    });
+  }
+
   private void writeBuffer(Stream stream, ByteBuf buff, boolean end, Promise<Void> listener) {
     writeToChannel(new MessageWrite() {
       @Override
@@ -669,6 +695,14 @@ public Future<Void> writeHead(io.vertx.core.http.impl.HttpRequestHead request, b
       return promise.future();
     }
 
+    @Override
+    public Future<Void> writeHeaders(MultiMap headers, boolean end) {
+      // HTTP/1 trailers ride on the terminating zero-length chunk.
+      Promise<Void> listener = context.promise();
+      conn.writeTrailers(this, (io.netty.handler.codec.http.HttpHeaders) headers, listener);
+      return listener.future();
+    }
+
     @Override
     public Future<Void> writeChunk(Buffer buff, boolean end) {
       if (buff != null || end) {