Adapt netty quic transport implementation to the latest change in transport API and SIGNIFICANTLY rework it, improving the correctness of the implementation

Author: whyoleg

rsocket-transports/netty-quic/src/jvmMain/kotlin/io/rsocket/kotlin/transport/netty/quic/NettyQuicClientTransport.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -59,7 +59,7 @@ private class NettyQuicClientTransportBuilderImpl : NettyQuicClientTransportBuil
     private var bootstrap: (Bootstrap.() -> Unit)? = null
     private var codec: (QuicClientCodecBuilder.() -> Unit)? = null
     private var ssl: (QuicSslContextBuilder.() -> Unit)? = null
-    private var quicBootstrap: (QuicChannelBootstrap.() -> Unit)? = null
+    private var quicBootstrap: QuicChannelBootstrap.() -> Unit = { }
 
     override fun channel(cls: KClass<out DatagramChannel>) {
         this.channelFactory = ReflectiveChannelFactory(cls.java)
@@ -114,24 +114,18 @@ private class NettyQuicClientTransportBuilderImpl : NettyQuicClientTransportBuil
         return NettyQuicClientTransportImpl(
             coroutineContext = context.supervisorContext() + bootstrap.config().group().asCoroutineDispatcher(),
             bootstrap = bootstrap,
-            quicBootstrap = quicBootstrap,
-            manageBootstrap = manageEventLoopGroup
-        )
+            quicBootstrap = quicBootstrap
+        ).also {
+            if (manageEventLoopGroup) it.shutdownOnCancellation(bootstrap.config().group())
+        }
     }
 }
 
 private class NettyQuicClientTransportImpl(
     override val coroutineContext: CoroutineContext,
     private val bootstrap: Bootstrap,
-    private val quicBootstrap: (QuicChannelBootstrap.() -> Unit)?,
-    manageBootstrap: Boolean,
+    private val quicBootstrap: QuicChannelBootstrap.() -> Unit,
 ) : NettyQuicClientTransport {
-    init {
-        if (manageBootstrap) callOnCancellation {
-            bootstrap.config().group().shutdownGracefully().awaitFuture()
-        }
-    }
-
     override fun target(remoteAddress: InetSocketAddress): NettyQuicClientTargetImpl = NettyQuicClientTargetImpl(
         coroutineContext = coroutineContext.supervisorContext(),
         bootstrap = bootstrap,
@@ -146,14 +140,29 @@ private class NettyQuicClientTransportImpl(
 private class NettyQuicClientTargetImpl(
     override val coroutineContext: CoroutineContext,
     private val bootstrap: Bootstrap,
-    private val quicBootstrap: (QuicChannelBootstrap.() -> Unit)?,
+    private val quicBootstrap: QuicChannelBootstrap.() -> Unit,
     private val remoteAddress: SocketAddress,
 ) : RSocketClientTarget {
     @RSocketTransportApi
-    override fun connectClient(handler: RSocketConnectionHandler): Job = launch {
-        QuicChannel.newBootstrap(bootstrap.bind().awaitChannel()).also { quicBootstrap?.invoke(it) }
+    override suspend fun connectClient(): RSocketConnection {
+        currentCoroutineContext().ensureActive()
+        coroutineContext.ensureActive()
+
+        val channel = QuicChannel.newBootstrap(
+            bootstrap.bind().awaitChannel()
+        )
+            .apply(quicBootstrap)
             .handler(
-                NettyQuicConnectionInitializer(handler, coroutineContext, isClient = true)
-            ).remoteAddress(remoteAddress).connect().awaitFuture()
+                NettyQuicConnectionInitializer(
+                    parentContext = coroutineContext,
+                    onConnection = null
+                )
+            )
+            .streamHandler(NettyQuicStreamInitializer)
+            .remoteAddress(remoteAddress)
+            .connect()
+            .awaitFuture()
+
+        return channel.attr(NettyQuicConnection.ATTRIBUTE).get()
     }
 }

rsocket-transports/netty-quic/src/jvmMain/kotlin/io/rsocket/kotlin/transport/netty/quic/NettyQuicConnection.kt

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2015-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.kotlin.transport.netty.quic
+
+import io.netty.channel.*
+import io.netty.incubator.codec.quic.*
+import io.netty.util.*
+import io.rsocket.kotlin.internal.io.*
+import io.rsocket.kotlin.transport.*
+import io.rsocket.kotlin.transport.netty.internal.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlin.coroutines.*
+
+@RSocketTransportApi
+internal class NettyQuicConnection(
+    parentContext: CoroutineContext,
+    private val channel: QuicChannel,
+    private val isClient: Boolean,
+) : RSocketMultiplexedConnection, ChannelInboundHandlerAdapter() {
+    override val coroutineContext: CoroutineContext = parentContext.childContext() + channel.eventLoop().asCoroutineDispatcher()
+    private val streamsContext = coroutineContext.supervisorContext()
+
+    private val inboundStreams = Channel<RSocketMultiplexedConnection.Stream>(Channel.UNLIMITED) {
+        it.cancel("Connection closed")
+    }
+
+    init {
+        @OptIn(DelicateCoroutinesApi::class)
+        launch(start = CoroutineStart.ATOMIC) {
+            try {
+                awaitCancellation()
+            } finally {
+                nonCancellable {
+                    inboundStreams.cancel()
+                    // stop streams first
+                    streamsContext.job.cancelAndJoin()
+                    channel.close().awaitFuture()
+                    // close UDP channel
+                    if (isClient) channel.parent().close().awaitFuture()
+                }
+            }
+        }
+    }
+
+    fun initStreamChannel(streamChannel: QuicStreamChannel) {
+        val stream = NettyQuicStream(streamsContext, streamChannel)
+        streamChannel.attr(NettyQuicStream.ATTRIBUTE).set(stream)
+        streamChannel.pipeline().addLast("rsocket-quic-stream", stream)
+
+        if (streamChannel.isLocalCreated) return
+
+        if (inboundStreams.trySend(stream).isFailure) stream.cancel("Connection closed")
+    }
+
+    override fun channelInactive(ctx: ChannelHandlerContext) {
+        cancel("Channel is not active")
+        ctx.fireChannelInactive()
+    }
+
+    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable?) {
+        cancel("exceptionCaught", cause)
+    }
+
+    override suspend fun createStream(): RSocketMultiplexedConnection.Stream {
+        val streamChannel = channel.createStream(QuicStreamType.BIDIRECTIONAL, NettyQuicStreamInitializer).awaitFuture()
+        return streamChannel.attr(NettyQuicStream.ATTRIBUTE).get()
+    }
+
+    override suspend fun acceptStream(): RSocketMultiplexedConnection.Stream? {
+        return inboundStreams.receiveCatching().getOrNull()
+    }
+
+    companion object {
+        val ATTRIBUTE: AttributeKey<NettyQuicConnection> = AttributeKey.newInstance<NettyQuicConnection>("rsocket-quic-connection")
+    }
+}
+
+@RSocketTransportApi
+internal class NettyQuicConnectionInitializer(
+    private val parentContext: CoroutineContext,
+    private val onConnection: ((RSocketConnection) -> Unit)?,
+) : ChannelInitializer<QuicChannel>() {
+    override fun initChannel(channel: QuicChannel) {
+        val connection = NettyQuicConnection(
+            parentContext = parentContext,
+            channel = channel,
+            isClient = onConnection == null
+        )
+        channel.attr(NettyQuicConnection.ATTRIBUTE).set(connection)
+
+        channel.pipeline().addLast(
+            "rsocket-connection",
+            connection
+        )
+
+        onConnection?.invoke(connection)
+    }
+}