Add QuicStream#abort that sends a STOP_SENDING frame.

Author: vietj

vertx-core/src/main/java/io/vertx/core/net/QuicStream.java

@@ -58,13 +58,23 @@ public interface QuicStream extends Socket {
   QuicStream resetHandler(@Nullable Handler<Integer> handler);
 
   /**
-   * Send a stream reset frame to the remote stream.
+   * Abruptly terminate the sending part of the stream with the specified application protocol {@code error} code
+   * argument, a {@code RESET} frame is sent to the remote peer.
    *
-   * @param error the error code
+   * @param error the application protocol error code
    * @return a future completed when the reset frame has been sent
    */
   Future<Void> reset(int error);
 
+  /**
+   * Instruct the remote peer that this part of the stream is no longer interested in received data and
+   * wants it to cease transmission, a {@code STOP_SENDING} frame is sent to the remote peer.
+   *
+   * @param error the error code
+   * @return a future completed when the {@code STOP_SENDING} frame has been sent
+   */
+  Future<Void> abort(int error);
+
   /**
    * Set a handler called when the stream is closed.
    *

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicStreamImpl.java

@@ -72,6 +72,14 @@ public Future<Void> reset(int error) {
     return promise.future();
   }
 
+  @Override
+  public Future<Void> abort(int error) {
+    PromiseInternal<Void> promise = context.promise();
+    ChannelFuture shutdownPromise = channel.shutdownInput(error);
+    shutdownPromise.addListener(promise);
+    return promise.future();
+  }
+
   @Override
   public NetworkMetrics<?> metrics() {
     return streamMetrics;

vertx-core/src/test/java/io/vertx/tests/net/quic/QuicServerTest.java

@@ -11,11 +11,14 @@
 package io.vertx.tests.net.quic;
 
 import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ChannelOutputShutdownException;
 import io.netty.handler.codec.quic.QuicClosedChannelException;
 import io.netty.util.NetUtil;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
+import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.internal.quic.QuicConnectionInternal;
@@ -40,6 +43,7 @@
 import java.security.KeyStore;
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -65,6 +69,10 @@ static QuicServerOptions serverOptions() {
     options.getTransportOptions().setInitialMaxStreamsBidirectional(100L);
     options.getTransportOptions().setInitialMaxStreamsUnidirectional(100L);
     options.getTransportOptions().setActiveMigration(true);
+
+//    options.setClientAddressValidation(QuicClientAddressValidation.NONE);
+//    options.setKeyLogFile("/Users/julien/keylogfile.txt");
+
     return options;
   }
 
@@ -381,6 +389,82 @@ public void testClientReset() throws Exception {
     }
   }
 
+  @Test
+  public void testClientAbortReading() throws Exception {
+    waitFor(3);
+    disableThreadChecks();
+    QuicServer server = QuicServer.create(vertx, serverOptions());
+    AtomicReference<QuicStream> streamRef = new AtomicReference<>();
+    server.handler(conn -> {
+      conn.streamHandler(stream -> {
+        streamRef.set(stream);
+        stream.handler(buff -> {
+          stream.write(buff);
+        });
+      });
+    });
+    server.bind(SocketAddress.inetSocketAddress(9999, "localhost")).await();
+    QuicTestClient client = new QuicTestClient(new NioEventLoopGroup(1));
+    try {
+      client = new QuicTestClient(new NioEventLoopGroup(1));
+      QuicTestClient.Connection connection = client.connect(new InetSocketAddress(NetUtil.LOCALHOST4, 9999));
+      QuicTestClient.Stream stream = connection.newStream();
+      stream.create();
+      AtomicInteger received = new AtomicInteger();
+      stream.handler(buff -> received.addAndGet(buff.length));
+      stream.write("ping");
+      assertWaitUntil(() -> received.get() > 0);
+      stream.abort(10);
+      try {
+        streamRef.get().write("test").await();
+        fail();
+      } catch (Exception e) {
+        assertEquals(ChannelOutputShutdownException.class, e.getClass());
+        assertEquals("STOP_SENDING frame received", e.getMessage());
+      }
+    } finally {
+      client.close();
+      server.close().await();
+    }
+  }
+
+  @Test
+  public void testServerAbortReading() throws Exception {
+    waitFor(3);
+    disableThreadChecks();
+    QuicServer server = QuicServer.create(vertx, serverOptions());
+    Promise<Void> writeLatch = Promise.promise();
+    server.handler(conn -> {
+      conn.streamHandler(stream -> {
+        stream.handler(buff -> {
+          stream
+            .abort(10)
+            .onComplete(writeLatch);
+        });
+      });
+    });
+    server.bind(SocketAddress.inetSocketAddress(9999, "localhost")).await();
+    QuicTestClient client = new QuicTestClient(new NioEventLoopGroup(1));
+    try {
+      client = new QuicTestClient(new NioEventLoopGroup(1));
+      QuicTestClient.Connection connection = client.connect(new InetSocketAddress(NetUtil.LOCALHOST4, 9999));
+      QuicTestClient.Stream stream = connection.newStream();
+      stream.create();
+      stream.write("ping");
+      writeLatch.future().await();
+      try {
+        stream.streamChannel.writeAndFlush(Unpooled.copiedBuffer("test", StandardCharsets.UTF_8)).sync();
+        fail();
+      } catch (Exception e) {
+        assertSame(ChannelOutputShutdownException.class, e.getClass());
+        assertEquals("STOP_SENDING frame received", e.getMessage());
+      }
+    } finally {
+      client.close();
+      server.close().await();
+    }
+  }
+
   @Test
   public void testServerReset() throws Exception {
     disableThreadChecks();

vertx-core/src/test/java/io/vertx/tests/net/quic/QuicTestClient.java

@@ -219,6 +219,13 @@ public void reset(int errorCode) throws Exception {
       streamChannel.shutdownOutput(errorCode).sync();
     }
 
+    public void abort(int errorCode) throws Exception {
+      if (streamChannel == null) {
+        throw new IllegalStateException();
+      }
+      streamChannel.shutdownInput(errorCode).sync();
+    }
+
     public Stream handler(Consumer<byte[]> handler) {
       this.handler = handler;
       return this;