Improve HttpClientRequest reset support - fixes #1642

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

@@ -137,13 +137,13 @@ public interface HttpClientRequest extends WriteStream<Buffer>, ReadStream<HttpC
   /**
    * Set the request host.<p/>
    *
-   * For HTTP2 it sets the {@literal :authority} pseudo header otherwise it sets the {@literal Host} header
+   * For HTTP/2 it sets the {@literal :authority} pseudo header otherwise it sets the {@literal Host} header
    */
   @Fluent
   HttpClientRequest setHost(String host);
 
   /**
-   * @return the request host. For HTTP2 it returns the {@literal :authority} pseudo header otherwise it returns the {@literal Host} header
+   * @return the request host. For HTTP/2 it returns the {@literal :authority} pseudo header otherwise it returns the {@literal Host} header
    */
   String getHost();
 
@@ -308,17 +308,28 @@ public interface HttpClientRequest extends WriteStream<Buffer>, ReadStream<HttpC
 
   /**
    * Reset this stream with the error code {@code 0}.
+   *
+   * @see #reset(long)
    */
-  default void reset() {
-    reset(0L);
+  default boolean reset() {
+    return reset(0L);
   }
 
   /**
-   * Reset this stream with the error {@code code}.
+   * Reset this request:
+   * <p/>
+   * <ul>
+   *   <li>for HTTP/2, this performs send an HTTP/2 reset frame with the specified error {@code code}</li>
+   *   <li>for HTTP/1.x, this closes the connection after the current in-flight requests are ended</li>
+   * </ul>
+   * <p/>
+   * When the request has not yet been sent, the request will be aborted and false is returned as indicator.
+   * <p/>
    *
    * @param code the error code
+   * @return true when reset has been performed
    */
-  void reset(long code);
+  boolean reset(long code);
 
   /**
    * @return the {@link HttpConnection} associated with this request

src/main/java/io/vertx/core/http/impl/ClientConnection.java

@@ -44,6 +44,7 @@
 
 import java.net.URI;
 import java.util.ArrayDeque;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Queue;
@@ -76,7 +77,7 @@ class ClientConnection extends ConnectionBase implements HttpClientConnection, H
   private final Http1xPool pool;
   private final Object endpointMetric;
   // Requests can be pipelined so we need a queue to keep track of requests
-  private final Queue<HttpClientRequestImpl> requests = new ArrayDeque<>();
+  private final Deque<HttpClientRequestImpl> requests = new ArrayDeque<>();
   private final Object metric;
   private final HttpClientMetrics metrics;
   private final HttpVersion version;
@@ -87,6 +88,7 @@ class ClientConnection extends ConnectionBase implements HttpClientConnection, H
   private HttpClientRequestImpl requestForResponse;
   private WebSocketImpl ws;
 
+  private boolean reset;
   private boolean paused;
   private Buffer pausedChunk;
 
@@ -377,7 +379,18 @@ void handleResponseEnd(LastHttpContent trailer) {
         // currently Vertx forces the Connection header if keepalive is enabled for 1.0
         close = true;
       }
-      pool.responseEnded(this, close);
+
+      if (close) {
+        pool.responseEnded(this, true);
+      } else {
+        if (reset) {
+          if (requests.isEmpty()) {
+            pool.responseEnded(this, true);
+          }
+        } else {
+          pool.responseEnded(this, false);
+        }
+      }
     }
     currentResponse = null;
   }
@@ -426,7 +439,17 @@ public ContextImpl getContext() {
   }
 
   public void reset(long code) {
-    throw new UnsupportedOperationException("HTTP/1.x request cannot be reset");
+    if (currentRequest == null) {
+      throw new IllegalStateException();
+    }
+    if (!reset) {
+      currentRequest = null;
+      reset = true;
+      requests.removeLast();
+      if (requests.size() == 0) {
+        pool.responseEnded(this, true);
+      }
+    }
   }
 
   private HttpRequest createRequest(HttpVersion version, HttpMethod method, String rawMethod, String uri, MultiMap headers) {

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

@@ -362,13 +362,19 @@ public HttpClientRequest pushHandler(Handler<HttpClientRequest> handler) {
   }
 
   @Override
-  public void reset(long code) {
+  public boolean reset(long code) {
     synchronized (getLock()) {
-      if (stream == null) {
-        throw new IllegalStateException("Cannot reset the request that is not yet connected");
+      if (completed) {
+        throw new IllegalStateException("Request already completed");
       }
+      exceptionOccurred = true;
       completed = true;
-      stream.reset(code);
+      if (stream != null) {
+        stream.reset(code);
+        return true;
+      } else {
+        return false;
+      }
     }
   }
 

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

@@ -103,9 +103,10 @@ public HttpClientRequest connectionHandler(@Nullable Handler<HttpConnection> han
   }
 
   @Override
-  public void reset(long code) {
+  public boolean reset(long code) {
     synchronized (conn) {
       stream.reset(code);
+      return true;
     }
   }