Resetting a client stream should make the stream has ended

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

@@ -353,7 +353,14 @@ public void endRequest() {
 
     @Override
     public void reset(long code) {
-      writeReset(code);
+      if (!(requestEnded && responseEnded)) {
+        requestEnded = true;
+        responseEnded = true;
+        writeReset(code);
+        if (conn.metrics.isEnabled()) {
+          conn.metrics.requestReset(request.metric());
+        }
+      }
     }
 
     @Override

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

@@ -350,6 +350,7 @@ public void reset(long code) {
       if (stream == null) {
         throw new IllegalStateException("Cannot reset the request that is not yet connected");
       }
+      completed = true;
       stream.reset(code);
     }
   }

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

@@ -33,33 +33,33 @@
  */
 class HttpClientRequestPushPromise extends HttpClientRequestBase {
 
-  private final Http2ClientConnection handler;
-  private final Http2ClientConnection.Http2ClientStream clientStream;
+  private final Http2ClientConnection conn;
+  private final Http2ClientConnection.Http2ClientStream stream;
   private final HttpMethod method;
   private final String uri;
   private final String host;
   private final MultiMap headers;
   private Handler<HttpClientResponse> respHandler;
 
   public HttpClientRequestPushPromise(
-      Http2ClientConnection handler,
-      Http2Stream clientStream,
+      Http2ClientConnection conn,
+      Http2Stream stream,
       HttpClientImpl client,
       HttpMethod method,
       String uri,
       String host,
       MultiMap headers) throws Http2Exception {
     super(client, method, host, uri);
-    this.handler = handler;
-    this.clientStream = new Http2ClientConnection.Http2ClientStream(handler, this, clientStream);
+    this.conn = conn;
+    this.stream = new Http2ClientConnection.Http2ClientStream(conn, this, stream);
     this.method = method;
     this.uri = uri;
     this.host = host;
     this.headers = headers;
   }
 
   Http2ClientConnection.Http2ClientStream getStream() {
-    return clientStream;
+    return stream;
   }
 
   @Override
@@ -90,7 +90,7 @@ public HttpClientRequest handler(Handler<HttpClientResponse> handler) {
 
   @Override
   public HttpConnection connection() {
-    return handler;
+    return conn;
   }
 
   @Override
@@ -100,7 +100,9 @@ public HttpClientRequest connectionHandler(@Nullable Handler<HttpConnection> han
 
   @Override
   public void reset(long code) {
-    clientStream.reset(code);
+    synchronized (conn) {
+      stream.reset(code);
+    }
   }
 
   @Override

src/test/java/io/vertx/test/core/Http2ClientTest.java

@@ -17,7 +17,6 @@
 package io.vertx.test.core;
 
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
@@ -78,6 +77,8 @@
 import java.util.function.BiFunction;
 import java.util.zip.GZIPOutputStream;
 
+import static io.vertx.test.core.TestUtils.assertIllegalStateException;
+
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
@@ -652,7 +653,7 @@ public void testServerResetClientStreamDuringResponse() throws Exception {
   }
 
   @Test
-  public void testClientResetServerStream() throws Exception {
+  public void testClientResetServerStreamDuringRequest() throws Exception {
     Future<Void> bufReceived = Future.future();
     server.requestHandler(req -> {
       req.handler(buf -> {
@@ -677,6 +678,30 @@ public void testClientResetServerStream() throws Exception {
     await();
   }
 
+  @Test
+  public void testClientResetServerStreamDuringResponse() throws Exception {
+    server.requestHandler(req -> {
+      req.exceptionHandler(err -> {
+        assertTrue(err instanceof StreamResetException);
+      });
+      req.response().exceptionHandler(err -> {
+        assertTrue(err instanceof StreamResetException);
+        assertEquals(10L, ((StreamResetException) err).getCode());
+        testComplete();
+      });
+      req.response().setChunked(true).write(Buffer.buffer("some-data"));
+    });
+    startServer();
+    HttpClientRequest req = client.get(DEFAULT_HTTPS_PORT, DEFAULT_HTTPS_HOST, "/somepath");
+    req.handler(resp -> {
+      resp.exceptionHandler(this::fail);
+      req.reset(10);
+      assertIllegalStateException(() -> req.write(Buffer.buffer()));
+      assertIllegalStateException(req::end);
+    }).setChunked(true).write(Buffer.buffer("hello"));
+    await();
+  }
+
   @Test
   public void testPushPromise() throws Exception {
     waitFor(2);