Repository: eclipse-vertx/vert.x

Problem statement:
HTTP/1.1 client metric is reset for allocated but non sent requests
The client calls the HTTP client metrics SPI for HTTP/1.1 requests which have been allocated but not yet sent: every request that is created by `HttpClient#request` that has not yet been sent calls back the SPI when it should not.


Hints:

Patch:
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
@@ -1040,17 +1040,6 @@ public  void handleInterestedOpsChanged() {
     context.execute(writable, handler);
   }
 
-  /**
-   * @return a list of all pending streams
-   */
-  private Iterable<Stream> pendingStreams() {
-    // There might be duplicate between the requets list and the responses list
-    LinkedHashSet<Stream> list = new LinkedHashSet<>();
-    list.addAll(requests);
-    list.addAll(responses);
-    return list;
-  }
-
   protected void handleClosed() {
     super.handleClosed();
     long timerID = shutdownTimerID;
@@ -1071,15 +1060,21 @@ protected void handleClosed() {
     }
     WebSocketImpl ws;
     VertxTracer tracer = context.tracer();
-    Iterable<Stream> streams;
+    List<Stream> allocatedStreams;
+    List<Stream> sentStreams;
     synchronized (this) {
       ws = webSocket;
-      streams = pendingStreams();
+      sentStreams = new ArrayList<>(responses);
+      allocatedStreams = new ArrayList<>(requests);
+      allocatedStreams.removeAll(responses);
     }
     if (ws != null) {
       ws.handleConnectionClosed();
     }
-    for (Stream stream : streams) {
+    for (Stream stream : allocatedStreams) {
+      stream.context.execute(null, v -> stream.handleClosed());
+    }
+    for (Stream stream : sentStreams) {
       if (metrics != null) {
         metrics.requestReset(stream.metric);
       }
@@ -1104,15 +1099,16 @@ protected void handleIdle() {
   protected void handleException(Throwable e) {
     super.handleException(e);
     WebSocketImpl ws;
-    Iterable<Stream> streams;
+    LinkedHashSet<Stream> allStreams = new LinkedHashSet<>();
     synchronized (this) {
       ws = webSocket;
-      streams = pendingStreams();
+      allStreams.addAll(requests);
+      allStreams.addAll(responses);
     }
     if (ws != null) {
       ws.handleException(e);
     }
-    for (Stream stream : streams) {
+    for (Stream stream : allStreams) {
       stream.handleException(e);
     }
   }


Test patch:
diff --git a/src/test/java/io/vertx/core/http/Http1xMetricsTest.java b/src/test/java/io/vertx/core/http/Http1xMetricsTest.java
--- a/src/test/java/io/vertx/core/http/Http1xMetricsTest.java
+++ b/src/test/java/io/vertx/core/http/Http1xMetricsTest.java
@@ -10,10 +10,31 @@
  */
 package io.vertx.core.http;
 
+import io.vertx.test.fakemetrics.FakeHttpClientMetrics;
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+
 public class Http1xMetricsTest extends HttpMetricsTestBase {
 
   public Http1xMetricsTest() {
     super(HttpVersion.HTTP_1_1);
   }
 
+  @Test
+  public void testAllocatedStreamResetShouldNotCallMetricsLifecycle() throws Exception {
+    server.requestHandler(req -> {
+      fail();
+    });
+    startServer();
+    CountDownLatch latch = new CountDownLatch(1);
+    client = vertx.createHttpClient(createBaseClientOptions().setIdleTimeout(2));
+    client.request(requestOptions).onComplete(onSuccess(req -> {
+      req.exceptionHandler(err -> {
+        latch.countDown();
+      });
+      req.connection().close();
+    }));
+    awaitLatch(latch);
+  }
 }
diff --git a/src/test/java/io/vertx/test/fakemetrics/FakeHttpClientMetrics.java b/src/test/java/io/vertx/test/fakemetrics/FakeHttpClientMetrics.java
--- a/src/test/java/io/vertx/test/fakemetrics/FakeHttpClientMetrics.java
+++ b/src/test/java/io/vertx/test/fakemetrics/FakeHttpClientMetrics.java
@@ -155,6 +155,7 @@ public static void sanityCheck() {
     if (err != null) {
       AssertionFailedError afe = new AssertionFailedError();
       afe.initCause(err);
+      unexpectedError = null;
       throw afe;
     }
   }