Repository: eclipse-vertx/vert.x

Problem statement:
HTTP/1.x client connection shutdown
HTTP/2 supports a graceful connection shutdown that relies on go away frames. We can provide a degraded support for HTTP/1.x for client connection that will have a similar effect since the client initiates the requests.

A HTTP/1.x client connection shutdown evicts the connection from the pool and waits until all current requests are terminated to close the connection.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/http/HttpConnection.java b/src/main/java/io/vertx/core/http/HttpConnection.java
--- a/src/main/java/io/vertx/core/http/HttpConnection.java
+++ b/src/main/java/io/vertx/core/http/HttpConnection.java
@@ -116,21 +116,25 @@ default HttpConnection goAway(long errorCode, int lastStreamId) {
   HttpConnection shutdownHandler(@Nullable  Handler<Void> handler);
 
   /**
-   * Initiate a connection shutdown, a go away frame is sent and the connection is closed when all current active streams
-   * are closed or after a time out of 30 seconds.
-   * <p/>
-   * This is not implemented for HTTP/1.x.
+   * Initiate a graceful connection shutdown, the connection is taken out of service and closed when all current requests
+   * are processed, otherwise after 30 seconds the connection will be closed. Client connection are immediately removed
+   * from the pool.
+   *
+   * <ul>
+   *   <li>HTTP/2 connections will send a go away frame immediately to signal the other side the connection will close</li>
+   *   <li>HTTP/1.x client connection supports this feature</li>
+   *   <li>HTTP/1.x server connections do not support this feature</li>
+   * </ul>
    *
    * @return a reference to this, so the API can be used fluently
    */
   @Fluent
-  HttpConnection shutdown();
+  default HttpConnection shutdown() {
+    return shutdown(30000L);
+  }
 
   /**
-   * Initiate a connection shutdown, a go away frame is sent and the connection is closed when all current streams
-   * will be closed or the {@code timeout} is fired.
-   * <p/>
-   * This is not implemented for HTTP/1.x.
+   * Like {@link #shutdown()} but with a configurable timeout value.
    *
    * @param timeoutMs the timeout in milliseconds
    * @return a reference to this, so the API can be used fluently
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
--- a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -73,6 +73,8 @@ class Http1xClientConnection extends Http1xConnectionBase<WebSocketImpl> impleme
   private StreamImpl responseInProgress;                         // The request waiting for a response
 
   private boolean close;
+  private long timerID;
+  private boolean shutdown;
   private boolean upgraded;
   private int keepAliveTimeout;
   private long expirationTimestamp;
@@ -95,6 +97,7 @@ class Http1xClientConnection extends Http1xConnectionBase<WebSocketImpl> impleme
     this.server = server;
     this.metrics = metrics;
     this.version = version;
+    this.timerID = -1L;
     this.endpointMetric = endpointMetric;
     this.keepAliveTimeout = options.getKeepAliveTimeout();
   }
@@ -374,7 +377,6 @@ public void reset(Throwable cause) {
       synchronized (conn) {
         if (conn.requestInProgress == this) {
           if (request == null) {
-            // Is that possible in practice ???
             conn.handleRequestEnd(true);
           } else {
             conn.close();
@@ -533,9 +535,15 @@ void handleException(Throwable cause) {
   }
 
   private void checkLifecycle() {
-    if (upgraded) {
-      // Do nothing
-    } else if (close) {
+    boolean close;
+    synchronized (this) {
+      if (upgraded) {
+        // Do nothing
+        return;
+      }
+      close = this.close;
+    }
+    if (close) {
       close();
     } else {
       recycle();
@@ -781,6 +789,10 @@ protected void handleClosed() {
     WebSocketImpl ws;
     List<StreamImpl> list = Collections.emptyList();
     synchronized (this) {
+      if (timerID > 0L) {
+        vertx.cancelTimer(timerID);
+        timerID = -1L;
+      }
       ws = this.ws;
       for (StreamImpl r = responseInProgress;r != null;r = r.next) {
         if (metrics != null) {
@@ -841,7 +853,39 @@ public boolean isValid() {
   }
 
   private void recycle() {
-    expirationTimestamp = keepAliveTimeout == 0 ? 0L : System.currentTimeMillis() + keepAliveTimeout * 1000;
-    listener.onRecycle();
+    if (shutdown) {
+      if (requestInProgress == null && responseInProgress == null) {
+        close();
+      }
+    } else {
+      expirationTimestamp = keepAliveTimeout == 0 ? 0L : System.currentTimeMillis() + keepAliveTimeout * 1000;
+      listener.onRecycle();
+    }
+  }
+
+  @Override
+  public HttpConnection shutdown(long timeoutMs) {
+    synchronized (this) {
+      if (upgraded) {
+        throw new IllegalStateException();
+      }
+      if (shutdown) {
+        return this;
+      }
+      if (timeoutMs > 0) {
+        timerID = vertx.setTimer(timeoutMs, id -> {
+          synchronized (Http1xClientConnection.this) {
+            timerID = -1L;
+          }
+          close();
+        });
+      } else {
+        close = true;
+      }
+      shutdown = true;
+    }
+    listener.onEvict();
+    checkLifecycle();
+    return this;
   }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xConnectionBase.java b/src/main/java/io/vertx/core/http/impl/Http1xConnectionBase.java
--- a/src/main/java/io/vertx/core/http/impl/Http1xConnectionBase.java
+++ b/src/main/java/io/vertx/core/http/impl/Http1xConnectionBase.java
@@ -189,11 +189,6 @@ public HttpConnection shutdownHandler(@Nullable Handler<Void> handler) {
     throw new UnsupportedOperationException("HTTP/1.x connections don't support GOAWAY");
   }
 
-  @Override
-  public HttpConnection shutdown() {
-    throw new UnsupportedOperationException("HTTP/1.x connections don't support GOAWAY");
-  }
-
   @Override
   public HttpConnection shutdown(long timeoutMs) {
     throw new UnsupportedOperationException("HTTP/1.x connections don't support GOAWAY");
diff --git a/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java b/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
--- a/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
+++ b/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
@@ -388,11 +388,6 @@ public HttpConnection shutdown(long timeout) {
     return this;
   }
 
-  @Override
-  public HttpConnection shutdown() {
-    return shutdown(30000);
-  }
-
   @Override
   public Http2ConnectionBase closeHandler(Handler<Void> handler) {
     return (Http2ConnectionBase) super.closeHandler(handler);
diff --git a/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java
--- a/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java
+++ b/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java
@@ -423,11 +423,6 @@ public HttpConnection goAway(long errorCode, int lastStreamId, Buffer debugData)
     return current.goAway(errorCode, lastStreamId, debugData);
   }
 
-  @Override
-  public HttpConnection shutdown() {
-    return current.shutdown();
-  }
-
   @Override
   public HttpConnection shutdown(long timeoutMs) {
     return current.shutdown(timeoutMs);


Test patch:
diff --git a/src/test/java/io/vertx/core/http/Http1xTest.java b/src/test/java/io/vertx/core/http/Http1xTest.java
--- a/src/test/java/io/vertx/core/http/Http1xTest.java
+++ b/src/test/java/io/vertx/core/http/Http1xTest.java
@@ -4748,4 +4748,116 @@ public void testAsyncPipelinedRequestDequeuing() throws Exception {
     }
     await();
   }
+
+  @Test
+  public void testClientConnectionGracefulShutdown() throws Exception {
+    int numReq = 3;
+    AtomicReference<HttpConnection> clientConnection = new AtomicReference<>();
+    AtomicInteger count = new AtomicInteger();
+    server.requestHandler(req -> {
+      if (count.getAndIncrement() == 0) {
+        clientConnection.get().shutdown();
+      }
+      req.response().end();
+    });
+    startServer(testAddress);
+    AtomicInteger responses = new AtomicInteger();
+    client = vertx.createHttpClient(createBaseClientOptions().setMaxPoolSize(1).setPipelining(true));
+    client.connectionHandler(conn -> {
+      conn.closeHandler(v -> {
+        assertEquals(3, responses.get());
+        testComplete();
+      });
+      clientConnection.set(conn);
+    });
+    for (int i = 0;i < numReq;i++) {
+      client.request(HttpMethod.PUT, testAddress, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+        responses.incrementAndGet();
+      }).end();
+    }
+    await();
+  }
+
+  @Test
+  public void testClientConnectionGracefulShutdownWhenRequestCompletedAfterResponse() throws Exception {
+    server.requestHandler(req -> {
+      req.response().end();
+    });
+    startServer(testAddress);
+    client.request(HttpMethod.PUT, testAddress, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+      AtomicBoolean requestEnded = new AtomicBoolean();
+      HttpClientRequest req = resp.request();
+      HttpConnection conn = req.connection();
+      conn.closeHandler(v -> {
+        assertTrue(requestEnded.get());
+        testComplete();
+      });
+      conn.shutdown();
+      resp.endHandler(v -> {
+        vertx.runOnContext(v2 -> {
+          requestEnded.set(true);
+          req.end();
+        });
+      });
+    }).setChunked(true).sendHead();
+    await();
+  }
+
+  @Test
+  public void testClientConnectionShutdownTimedOut() throws Exception {
+    AtomicReference<HttpConnection> clientConnectionRef = new AtomicReference<>();
+    int numReq = 3;
+    waitFor(numReq + 1);
+    server.requestHandler(req -> {
+      HttpConnection clientConnection = clientConnectionRef.getAndSet(null);
+      if (clientConnection != null) {
+        long now = System.currentTimeMillis();
+        clientConnection.closeHandler(v -> {
+          assertTrue(System.currentTimeMillis() - now >= 500L);
+          complete();
+        });
+        clientConnection.shutdown(500L);
+      }
+    });
+    startServer(testAddress);
+    client = vertx.createHttpClient(createBaseClientOptions().setMaxPoolSize(1).setPipelining(true));
+    client.connectionHandler(clientConnectionRef::set);
+    for (int i = 0;i < numReq;i++) {
+      AtomicBoolean failed = new AtomicBoolean();
+      client.request(HttpMethod.PUT, testAddress, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {
+      }).exceptionHandler(err -> {
+        if (failed.compareAndSet(false, true)) {
+          complete();
+        }
+      }).end();
+    }
+    await();
+  }
+
+  @Test
+  public void testClientConnectionShutdownNow() throws Exception {
+    AtomicReference<HttpConnection> clientConnectionRef = new AtomicReference<>();
+    waitFor(2);
+    server.requestHandler(req -> {
+      long now = System.currentTimeMillis();
+      HttpConnection clientConnection = clientConnectionRef.get();
+      clientConnection.closeHandler(v -> {
+        assertTrue(System.currentTimeMillis() - now <= 2000L);
+        complete();
+      });
+      clientConnection.shutdown(0L);
+    });
+    startServer(testAddress);
+    client = vertx.createHttpClient(createBaseClientOptions().setMaxPoolSize(1).setPipelining(true));
+    client.connectionHandler(clientConnectionRef::set);
+    AtomicBoolean failed = new AtomicBoolean();
+    client
+      .request(HttpMethod.PUT, testAddress, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI, resp -> {})
+      .exceptionHandler(err -> {
+        if (failed.compareAndSet(false, true)) {
+          complete();
+        }
+      }).end();
+    await();
+  }
 }