- Only keep track in the stream group of streams we can shutdown

- Add Quic stream channel initializer in internal connection
- Correctly name the connection channel handler so it can be referenced when updating the quic connection pipeline

Author: vietj

vertx-core/src/main/java/io/vertx/core/internal/quic/QuicConnectionInternal.java

@@ -11,13 +11,27 @@
 package io.vertx.core.internal.quic;
 
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.handler.codec.quic.QuicStreamChannel;
+import io.vertx.core.Future;
+import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.net.QuicConnection;
+import io.vertx.core.net.QuicStream;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
 public interface QuicConnectionInternal extends QuicConnection {
 
+  ContextInternal context();
+
+  Future<QuicStream> createStream(boolean bidirectional, Function<Consumer<QuicStreamChannel>, ChannelInitializer<QuicStreamChannel>> blah);
+
   ChannelHandlerContext channelHandlerContext();
 
+  Future<Void> closeFuture();
+
 }

vertx-core/src/main/java/io/vertx/core/internal/quic/QuicStreamInternal.java

@@ -13,6 +13,7 @@
 import io.vertx.codegen.annotations.Nullable;
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.net.SocketInternal;
 import io.vertx.core.net.QuicStream;
 
@@ -21,6 +22,8 @@
  */
 public interface QuicStreamInternal extends QuicStream, SocketInternal {
 
+  ContextInternal context();
+
   @Override
   QuicStreamInternal messageHandler(Handler<Object> handler);
 

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

@@ -24,6 +24,8 @@
 @VertxGen
 public interface QuicStream extends Socket {
 
+  long id();
+
   /**
    * @return whether the stream is unidirectional or bidirectional
    */

vertx-core/src/main/java/io/vertx/core/net/impl/VertxHandler.java

@@ -39,6 +39,23 @@ private VertxHandler(Function<ChannelHandlerContext, C> connectionFactory) {
     this.connectionFactory = connectionFactory;
   }
 
+  public static ByteBuf copyBuffer(ByteBuf byteBuf) {
+    Class<?> allocClass;
+    if (byteBuf != Unpooled.EMPTY_BUFFER &&
+      ((allocClass = byteBuf.alloc().getClass()) == AdaptiveByteBufAllocator.class
+        || allocClass == PooledByteBufAllocator.class
+        || byteBuf instanceof CompositeByteBuf)) {
+      if (byteBuf.isReadable()) {
+        ByteBuf buffer = VertxByteBufAllocator.DEFAULT.heapBuffer(byteBuf.readableBytes());
+        buffer.writeBytes(byteBuf, byteBuf.readerIndex(), byteBuf.readableBytes());
+        return buffer;
+      } else {
+        return Unpooled.EMPTY_BUFFER;
+      }
+    }
+    return byteBuf;
+  }
+
   /**
    * Pooled {@code byteBuf} are copied and released, otherwise it is returned as is.
    *

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

@@ -94,7 +94,7 @@ public Future<QuicConnection> connect(SocketAddress address, QLogConfig qLogConf
         protected void initChannel(Channel ch) {
           connectionGroup.add(ch);
           QuicConnectionHandler handler = new QuicConnectionHandler(context, metrics, promise::tryComplete);
-          ch.pipeline().addLast(handler);
+          ch.pipeline().addLast("handler", handler);
         }
       })
       .remoteAddress(new InetSocketAddress(address.host(), address.port()));

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

@@ -12,10 +12,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
+import io.netty.channel.*;
 import io.netty.handler.codec.quic.QuicChannel;
 import io.netty.handler.codec.quic.QuicStreamChannel;
 import io.netty.handler.codec.quic.QuicStreamType;
@@ -42,6 +39,9 @@
 import javax.net.ssl.SSLEngine;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
 
 /**
  * @author <a href="mailto:[email protected]">Julien Viet</a>
@@ -93,7 +93,10 @@ public NetworkMetrics<?> metrics() {
   }
 
   void handleStream(QuicStreamChannel streamChannel) {
-    streamGroup.add(streamChannel);
+    if (streamChannel.type() == QuicStreamType.BIDIRECTIONAL || streamChannel.isLocalCreated()) {
+      // Only consider stream we can end for shutdown, e.g. this excludes remote opened HTTP/3 control stream
+      streamGroup.add(streamChannel);
+    }
     VertxHandler<QuicStreamImpl> handler = VertxHandler.create(chctx -> new QuicStreamImpl(this, context, streamChannel, streamMetrics, chctx));
     handler.addHandler(stream -> {
       Handler<QuicStream> h = QuicConnectionImpl.this.handler;
@@ -163,22 +166,47 @@ public QuicConnection streamHandler(Handler<QuicStream> handler) {
 
   @Override
   public Future<QuicStream> createStream(boolean bidirectional) {
+    Function<Supplier<ChannelHandler>, ChannelInitializer<QuicStreamChannel>> blah = new Function<Supplier<ChannelHandler>, ChannelInitializer<QuicStreamChannel>>() {
+      @Override
+      public ChannelInitializer<QuicStreamChannel> apply(Supplier<ChannelHandler> channelHandlerSupplier) {
+        return new ChannelInitializer<>() {
+          @Override
+          protected void initChannel(QuicStreamChannel ch) throws Exception {
+            ChannelHandler abc = channelHandlerSupplier.get();
+            ch.pipeline().addLast("handler", abc);
+          }
+        };
+      }
+    };
+    Function<Consumer<QuicStreamChannel>, ChannelInitializer<QuicStreamChannel>> blah2 = new Function<Consumer<QuicStreamChannel>, ChannelInitializer<QuicStreamChannel>>() {
+      @Override
+      public ChannelInitializer<QuicStreamChannel> apply(Consumer<QuicStreamChannel> quicStreamChannelConsumer) {
+        return new ChannelInitializer<>() {
+          @Override
+          protected void initChannel(QuicStreamChannel ch) throws Exception {
+            quicStreamChannelConsumer.accept(ch);
+          }
+        };
+      }
+    };
+    return createStream(bidirectional, blah2);
+  }
+
+  @Override
+  public Future<QuicStream> createStream(boolean bidirectional, Function<Consumer<QuicStreamChannel>, ChannelInitializer<QuicStreamChannel>> blah) {
     // TODO : should use get or create context and test it ....
     Promise<QuicStream> promise = context.promise();
     VertxHandler<QuicStreamImpl> handler = VertxHandler.create(chctx -> new QuicStreamImpl(this, context, (QuicStreamChannel) chctx.channel(), streamMetrics, chctx));
     handler.addHandler(stream -> {
       promise.tryComplete(stream);
     });
     QuicStreamType type = bidirectional ? QuicStreamType.BIDIRECTIONAL : QuicStreamType.UNIDIRECTIONAL;
-    io.netty.util.concurrent.Future<QuicStreamChannel> future = channel.createStream(type, new ChannelInitializer<>() {
-      @Override
-      protected void initChannel(Channel ch) {
-        ch.pipeline().addLast("handler", handler);
-      }
+    ChannelInitializer<QuicStreamChannel> initializer = blah.apply(ch -> {
+      ch.pipeline().addLast("handler", handler);
     });
+    io.netty.util.concurrent.Future<QuicStreamChannel> future = channel.createStream(type, initializer);
     future.addListener(future1 -> {
       if (!future1.isSuccess()) {
-        future1.cause().printStackTrace(System.out);
         promise.tryFail(future1.cause());
       }
     });

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

@@ -128,7 +128,12 @@ private Future<Channel> bind(ContextInternal context, SocketAddress bindAddr, Ss
    Bootstrap bootstrap = new Bootstrap()
       .group(context.nettyEventLoop())
       .channelFactory(vertx.transport().datagramChannelFactory());
-    InetSocketAddress addr = new InetSocketAddress(bindAddr.hostName(), bindAddr.port());
+   InetSocketAddress addr;
+   if (bindAddr.hostAddress() != null) {
+     addr = new InetSocketAddress(bindAddr.hostAddress(), bindAddr.port());
+   } else {
+     addr = new InetSocketAddress(bindAddr.hostName(), bindAddr.port());
+   }
     ChannelHandler handler;
     try {
       handler = channelHandler(context, bindAddr, sslContextProvider, metrics);

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

@@ -146,7 +146,7 @@ protected void initChannel(Channel ch) {
                 QuicChannel channel = (QuicChannel) ch;
                 QuicConnectionHandler handler = new QuicConnectionHandler(context, metrics, QuicServerImpl.this.handler);
                 ChannelPipeline pipeline = channel.pipeline();
-                pipeline.addLast(handler);
+                pipeline.addLast("handler", handler);
               }
 
 /*

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

@@ -43,7 +43,7 @@ public class QuicStreamImpl extends SocketBase<QuicStreamImpl> implements QuicSt
   private final boolean localCreated;
   private Handler<Integer> resetHandler;
 
-  QuicStreamImpl(QuicConnection connection, ContextInternal context, QuicStreamChannel channel, NetworkMetrics<?> streamMetrics, ChannelHandlerContext chctx) {
+  public QuicStreamImpl(QuicConnection connection, ContextInternal context, QuicStreamChannel channel, NetworkMetrics<?> streamMetrics, ChannelHandlerContext chctx) {
     super(context, chctx);
     this.connection = connection;
     this.context = context;
@@ -53,6 +53,11 @@ public class QuicStreamImpl extends SocketBase<QuicStreamImpl> implements QuicSt
     this.localCreated = channel.isLocalCreated();
   }
 
+  @Override
+  public long id() {
+    return channel.streamId();
+  }
+
   @Override
   public QuicStream resetHandler(@Nullable Handler<Integer> handler) {
     this.resetHandler = handler;