Merge pull request #5747 from eclipse-vertx/http2-simplification

HTTP stream internal API

Author: vietj

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

@@ -16,6 +16,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.impl.HttpServerResponseImpl;
 import io.vertx.core.net.HostAndPort;
 import io.vertx.core.streams.ReadStream;
 import io.vertx.core.streams.WriteStream;
@@ -41,7 +42,7 @@
  * serving files from the server since buffers do not have to be read one by one
  * from the file and written to the outgoing socket. If the developer wants to use directly a
  * {@link java.nio.channels.FileChannel} and manage its lifecycle use {@link #sendFile(FileChannel)}.
- * This is not yet supported in HTTP/2 for {@link io.vertx.core.http.impl.Http2ServerResponse}.
+ * This is not yet supported in HTTP/2 for {@link HttpServerResponseImpl}.
  * <p>
  * It implements {@link io.vertx.core.streams.WriteStream} so it can be used with
  * {@link io.vertx.core.streams.Pipe} to pipe data with flow control.

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

@@ -37,7 +37,6 @@
 import io.vertx.core.http.HttpVersion;
 import io.vertx.core.http.WebSocketVersion;
 import io.vertx.core.http.impl.headers.HeadersAdaptor;
-import io.vertx.core.http.impl.http2.Http2ClientPush;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.PromiseInternal;
 import io.vertx.core.internal.buffer.BufferInternal;
@@ -378,6 +377,7 @@ private abstract static class Stream {
     protected final ContextInternal context;
     protected final int id;
 
+    private HttpVersion version;
     private Object trace;
     private Object metric;
     private HttpRequestHead request;
@@ -500,7 +500,7 @@ public boolean isWritable() {
     }
 
     @Override
-    public HttpClientStream headersHandler(Handler<HttpResponseHead> handler) {
+    public HttpClientStream headHandler(Handler<HttpResponseHead> handler) {
       this.headHandler = handler;
       return this;
     }
@@ -523,7 +523,7 @@ public HttpClientStream priorityChangeHandler(Handler<StreamPriority> handler) {
     }
 
     @Override
-    public HttpClientStream pushHandler(Handler<Http2ClientPush> handler) {
+    public HttpClientStream pushHandler(Handler<HttpClientPush> handler) {
       // No op
       return this;
     }
@@ -565,25 +565,25 @@ public ContextInternal context() {
     }
 
     @Override
-    public Future<Void> writeHead(HttpRequestHead request, boolean chunked, ByteBuf buf, boolean end, StreamPriority priority, boolean connect) {
+    public Future<Void> writeHead(HttpRequestHead request, boolean chunked, Buffer buf, boolean end, StreamPriority priority, boolean connect) {
       PromiseInternal<Void> promise = context.promise();
-      conn.writeHead(this, request, chunked, buf, end, connect, promise);
+      conn.writeHead(this, request, chunked, buf != null ? ((BufferInternal)buf).getByteBuf() : null, end, connect, promise);
       return promise.future();
     }
 
     @Override
-    public Future<Void> write(ByteBuf buff, boolean end) {
+    public Future<Void> writeChunk(Buffer buff, boolean end) {
       if (buff != null || end) {
         Promise<Void> listener = context.promise();
-        conn.writeBuffer(this, buff, end, listener);
+        conn.writeBuffer(this, buff != null ? ((BufferInternal)buff).getByteBuf() : null, end, listener);
         return listener.future();
       } else {
         throw new IllegalStateException("???");
       }
     }
 
     @Override
-    public Future<Void> writeFrame(int type, int flags, ByteBuf payload) {
+    public Future<Void> writeFrame(int type, int flags, Buffer payload) {
       throw new IllegalStateException("Cannot write an HTTP/2 frame over an HTTP/1.x connection");
     }
 
@@ -793,8 +793,7 @@ private void handleHttpMessage(HttpObject obj) {
       } else {
         version = io.vertx.core.http.HttpVersion.HTTP_1_1;
       }
-      handleResponseBegin(stream, new HttpResponseHead(
-        version,
+      handleResponseBegin(stream, version, new HttpResponseHead(
         response.status().code(),
         response.status().reasonPhrase(),
         new HeadersAdaptor(response.headers())));
@@ -822,7 +821,7 @@ private void handleChunk(ByteBuf chunk) {
     }
   }
 
-  private void handleResponseBegin(Stream stream, HttpResponseHead response) {
+  private void handleResponseBegin(Stream stream, HttpVersion version, HttpResponseHead response) {
     // How can we handle future undefined 1xx informational response codes?
     if (response.statusCode == HttpResponseStatus.CONTINUE.code()) {
       stream.handleContinue();
@@ -832,6 +831,7 @@ private void handleResponseBegin(Stream stream, HttpResponseHead response) {
       HttpRequestHead request;
       synchronized (this) {
         request = stream.request;
+        stream.version = version;
         stream.response = response;
         if (metrics != null) {
           metrics.responseBegin(stream.metric, response);
@@ -885,9 +885,11 @@ private void handleResponseChunk(Stream stream, ByteBuf chunk) {
 
   private void handleResponseEnd(Stream stream, LastHttpContent trailer) {
     boolean check;
-    HttpResponseHead response ;
+    HttpResponseHead response;
+    HttpVersion version;
     synchronized (this) {
       response = stream.response;
+      version = stream.version;
       if (response == null) {
         // 100-continue
         return;
@@ -903,7 +905,7 @@ private void handleResponseEnd(Stream stream, LastHttpContent trailer) {
         if (HttpHeaderValues.CLOSE.contentEqualsIgnoreCase(responseConnectionHeader) || HttpHeaderValues.CLOSE.contentEqualsIgnoreCase(requestConnectionHeader)) {
           // In all cases, if we have a close connection option then we SHOULD NOT treat the connection as persistent
           close = true;
-        } else if (response.version == HttpVersion.HTTP_1_0 && !HttpHeaderValues.KEEP_ALIVE.contentEqualsIgnoreCase(responseConnectionHeader)) {
+        } else if (version == HttpVersion.HTTP_1_0 && !HttpHeaderValues.KEEP_ALIVE.contentEqualsIgnoreCase(responseConnectionHeader)) {
           // In the HTTP/1.0 case both request/response need a keep-alive connection header the connection to be persistent
           // currently Vertx forces the Connection header if keepalive is enabled for 1.0
           close = true;

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

@@ -28,6 +28,7 @@
 import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.ServerWebSocketHandshake;
+import io.vertx.core.http.impl.headers.HeadersMultiMap;
 import io.vertx.core.internal.buffer.BufferInternal;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.HttpServerRequest;
@@ -114,6 +115,16 @@ protected void handleShutdown(ChannelPromise promise) {
     }
   }
 
+  @Override
+  public MultiMap newHeaders() {
+    return io.vertx.core.http.HttpHeaders.headers();
+  }
+
+  @Override
+  public boolean supportsSendFile() {
+    return true;
+  }
+
   TracingPolicy tracingPolicy() {
     return tracingPolicy;
   }

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

@@ -10,7 +10,6 @@
  */
 package io.vertx.core.http.impl;
 
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
@@ -23,7 +22,6 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.*;
 import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.impl.http2.Http2ClientPush;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.PromiseInternal;
 import io.vertx.core.internal.logging.Logger;
@@ -147,17 +145,17 @@ public ContextInternal context() {
     }
 
     @Override
-    public Future<Void> writeHead(HttpRequestHead request, boolean chunked, ByteBuf buf, boolean end, StreamPriority priority, boolean connect) {
+    public Future<Void> writeHead(HttpRequestHead request, boolean chunked, Buffer buf, boolean end, StreamPriority priority, boolean connect) {
       return delegate.writeHead(request, chunked, buf, end, priority, connect);
     }
 
     @Override
-    public Future<Void> write(ByteBuf buf, boolean end) {
-      return delegate.write(buf, end);
+    public Future<Void> writeChunk(Buffer buf, boolean end) {
+      return delegate.writeChunk(buf, end);
     }
 
     @Override
-    public Future<Void> writeFrame(int type, int flags, ByteBuf payload) {
+    public Future<Void> writeFrame(int type, int flags, Buffer payload) {
       return delegate.writeFrame(type, flags, payload);
     }
 
@@ -174,7 +172,7 @@ public HttpClientStream earlyHintsHandler(Handler<MultiMap> handler) {
     }
 
     @Override
-    public HttpClientStream pushHandler(Handler<Http2ClientPush> handler) {
+    public HttpClientStream pushHandler(Handler<HttpClientPush> handler) {
       delegate.pushHandler(handler);
       return this;
     }
@@ -186,8 +184,8 @@ public HttpClientStream customFrameHandler(Handler<HttpFrame> handler) {
     }
 
     @Override
-    public HttpClientStream headersHandler(Handler<HttpResponseHead> handler) {
-      delegate.headersHandler(handler);
+    public HttpClientStream headHandler(Handler<HttpResponseHead> handler) {
+      delegate.headHandler(handler);
       return this;
     }
 
@@ -281,7 +279,7 @@ private static class UpgradingStream implements HttpClientStream {
 
     void handleUpgrade(HttpClientConnection conn, HttpClientStream stream) {
       upgradedStream = stream;
-      upgradedStream.headersHandler(headHandler);
+      upgradedStream.headHandler(headHandler);
       upgradedStream.dataHandler(chunkHandler);
       upgradedStream.trailersHandler(trailersHandler);
       upgradedStream.priorityChangeHandler(priorityHandler);
@@ -293,7 +291,7 @@ void handleUpgrade(HttpClientConnection conn, HttpClientStream stream) {
       upgradedStream.pushHandler(pushHandler);
       upgradedStream.customFrameHandler(unknownFrameHandler);
       upgradedStream.closeHandler(closeHandler);
-      upgradingStream.headersHandler(null);
+      upgradingStream.headHandler(null);
       upgradingStream.dataHandler(null);
       upgradingStream.trailersHandler(null);
       upgradingStream.priorityChangeHandler(null);
@@ -352,7 +350,7 @@ void handleUpgrade(HttpClientConnection conn, HttpClientStream stream) {
     private Handler<Void> drainHandler;
     private Handler<Void> continueHandler;
     private Handler<MultiMap> earlyHintsHandler;
-    private Handler<Http2ClientPush> pushHandler;
+    private Handler<HttpClientPush> pushHandler;
     private Handler<HttpFrame> unknownFrameHandler;
     private Handler<Void> closeHandler;
 
@@ -373,11 +371,11 @@ public HttpClientConnection connection() {
      */
     @Override
     public Future<Void> writeHead(HttpRequestHead request,
-                          boolean chunked,
-                          ByteBuf buf,
-                          boolean end,
-                          StreamPriority priority,
-                          boolean connect) {
+                                  boolean chunked,
+                                  Buffer buf,
+                                  boolean end,
+                                  StreamPriority priority,
+                                  boolean connect) {
       UpgradeResult blah = new UpgradeResult() {
         @Override
         public void upgradeAccepted(HttpClientConnection connection, HttpClientStream upgradedStream) {
@@ -408,7 +406,7 @@ public void upgradeFailure(Throwable cause) {
 
     private void writeHead(HttpRequestHead head,
                            boolean chunked,
-                           ByteBuf buf,
+                           Buffer buf,
                            boolean end,
                            StreamPriority priority,
                            boolean connect,
@@ -477,7 +475,7 @@ public HttpClientStream earlyHintsHandler(Handler<MultiMap> handler) {
     }
 
     @Override
-    public HttpClientStream pushHandler(Handler<Http2ClientPush> handler) {
+    public HttpClientStream pushHandler(Handler<HttpClientPush> handler) {
       if (upgradedStream != null) {
         upgradedStream.pushHandler(handler);
       } else {
@@ -532,11 +530,11 @@ public UpgradingStream exceptionHandler(Handler<Throwable> handler) {
     }
 
     @Override
-    public HttpClientStream headersHandler(Handler<HttpResponseHead> handler) {
+    public HttpClientStream headHandler(Handler<HttpResponseHead> handler) {
       if (upgradedStream != null) {
-        upgradedStream.headersHandler(handler);
+        upgradedStream.headHandler(handler);
       } else {
-        upgradingStream.headersHandler(handler);
+        upgradingStream.headHandler(handler);
         headHandler = handler;
       }
       return this;
@@ -606,10 +604,10 @@ public boolean isWritable() {
     }
 
     @Override
-    public Future<Void> write(ByteBuf buf, boolean end) {
+    public Future<Void> writeChunk(Buffer buf, boolean end) {
       EventExecutor exec = upgradingConnection.channelHandlerContext().executor();
       if (exec.inEventLoop()) {
-        Future<Void> future = upgradingStream.write(buf, end);
+        Future<Void> future = upgradingStream.writeChunk(buf, end);
         if (end) {
           ChannelPipeline pipeline = upgradingConnection.channelHandlerContext().pipeline();
           future = future.andThen(ar -> {
@@ -622,15 +620,15 @@ public Future<Void> write(ByteBuf buf, boolean end) {
       } else {
         Promise<Void> promise = upgradingStream.context().promise();
         exec.execute(() -> {
-          Future<Void> future = write(buf, end);
+          Future<Void> future = writeChunk(buf, end);
           future.onComplete(promise);
         });
         return promise.future();
       }
     }
 
     @Override
-    public Future<Void> writeFrame(int type, int flags, ByteBuf payload) {
+    public Future<Void> writeFrame(int type, int flags, Buffer payload) {
       if (upgradedStream != null) {
         return upgradedStream.writeFrame(type, flags, payload);
       } else {
@@ -903,7 +901,7 @@ public interface Http2ChannelUpgrade {
 
     void upgrade(HttpClientStream upgradingStream,
                  HttpRequestHead request,
-                 ByteBuf content,
+                 Buffer content,
                  boolean end,
                  Channel channel,
                  boolean pooled,

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

@@ -259,7 +259,7 @@ private void http1xConnected(HttpVersion version,
           conn2.createStream(conn.context()).onComplete(ar -> {
             if (ar.succeeded()) {
               HttpClientStream stream = ar.result();
-              stream.headersHandler(resp -> {
+              stream.headHandler(resp -> {
                 Http2UpgradeClientConnection connection = (Http2UpgradeClientConnection) stream.connection();
                 HttpClientConnection unwrap = connection.unwrap();
                 future.tryComplete(unwrap);

vertx-core/src/main/java/io/vertx/core/http/impl/http2/Http2ClientPush.java renamed to vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientPush.java

@@ -8,40 +8,39 @@
  *
  * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
  */
-package io.vertx.core.http.impl.http2;
+package io.vertx.core.http.impl;
 
 import io.vertx.core.MultiMap;
 import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.impl.HttpClientStream;
 import io.vertx.core.net.HostAndPort;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.observability.HttpRequest;
 
 /**
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
-public class Http2ClientPush implements HttpRequest {
+public class HttpClientPush implements HttpRequest {
 
   private final String uri;
   private final HttpMethod method;
   private final HostAndPort authority;
   private final HttpClientStream stream;
   private final MultiMap headers;
 
-  Http2ClientPush(Http2HeadersMultiMap headers, HttpClientStream stream) {
+  public HttpClientPush(HttpRequestHead head, HttpClientStream stream) {
 
-    String rawMethod = headers.method().toString();
-    String authority = headers.authority() != null ? headers.authority().toString() : null;
+    String rawMethod = head.method().toString();
+    String authority = head.authority != null ? head.authority.toString() : null;
     int pos = authority == null ? -1 : authority.indexOf(':');
     if (pos == -1) {
       this.authority = HostAndPort.create(authority, 80);
     } else {
       this.authority = HostAndPort.create(authority.substring(0, pos), Integer.parseInt(authority.substring(pos + 1)));
     }
     this.method = HttpMethod.valueOf(rawMethod);
-    this.uri = headers.path().toString();
+    this.uri = head.uri;
     this.stream = stream;
-    this.headers = headers;
+    this.headers = head.headers;
   }
 
   public HttpClientStream stream() {
@@ -77,4 +76,5 @@ public String uri() {
   public HttpMethod method() {
     return method;
   }
+
 }