minor netty quic cleanup

Author: whyoleg

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

@@ -149,17 +149,20 @@ private class NettyQuicClientTargetImpl(
         coroutineContext.ensureActive()
 
         val channel = QuicChannel.newBootstrap(
-            bootstrap.clone()
-                .attr(ATTRIBUTE_TRANSPORT_CONTEXT, this@NettyQuicClientTargetImpl.coroutineContext)
-                .bind().awaitChannel()
+            bootstrap.bind().awaitChannel()
         )
             .apply(quicBootstrap)
-            .handler(NettyQuicConnectionInitializer)
+            .handler(
+                NettyQuicConnectionInitializer(
+                    parentContext = coroutineContext,
+                    onConnection = null
+                )
+            )
             .streamHandler(NettyQuicStreamInitializer)
             .remoteAddress(remoteAddress)
             .connect()
             .awaitFuture()
 
-        return channel.attr(ATTRIBUTE_CONNECTION).get()
+        return channel.attr(NettyQuicConnection.ATTRIBUTE).get()
     }
 }

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

@@ -18,6 +18,7 @@ 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.*
@@ -31,10 +32,11 @@ internal class NettyQuicConnection(
     private val channel: QuicChannel,
     private val isClient: Boolean,
 ) : RSocketMultiplexedConnection, ChannelInboundHandlerAdapter() {
-    private val inboundStreams = Channel<RSocketMultiplexedConnection.Stream>(Channel.UNLIMITED)
     override val coroutineContext: CoroutineContext = parentContext.childContext() + channel.eventLoop().asCoroutineDispatcher()
     private val streamsContext = coroutineContext.supervisorContext()
 
+    private val inboundStreams = Channel<RSocketMultiplexedConnection.Stream>(Channel.UNLIMITED)
+
     init {
         @OptIn(DelicateCoroutinesApi::class)
         launch(start = CoroutineStart.ATOMIC) {
@@ -55,7 +57,7 @@ internal class NettyQuicConnection(
 
     fun initStreamChannel(streamChannel: QuicStreamChannel) {
         val stream = NettyQuicStream(streamsContext, streamChannel)
-        streamChannel.attr(ATTRIBUTE_STREAM).set(stream)
+        streamChannel.attr(NettyQuicStream.ATTRIBUTE).set(stream)
         streamChannel.pipeline().addLast("rsocket-quic-stream", stream)
 
         if (streamChannel.isLocalCreated) return
@@ -74,40 +76,36 @@ internal class NettyQuicConnection(
 
     override suspend fun createStream(): RSocketMultiplexedConnection.Stream {
         val streamChannel = channel.createStream(QuicStreamType.BIDIRECTIONAL, NettyQuicStreamInitializer).awaitFuture()
-        return streamChannel.attr(ATTRIBUTE_STREAM).get()
+        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 object NettyQuicConnectionInitializer : ChannelInitializer<QuicChannel>() {
+internal class NettyQuicConnectionInitializer(
+    private val parentContext: CoroutineContext,
+    private val onConnection: ((RSocketConnection) -> Unit)?,
+) : ChannelInitializer<QuicChannel>() {
     override fun initChannel(channel: QuicChannel) {
-        val acceptor = channel.parent().attr(ATTRIBUTE_CONNECTION_ACCEPTOR).get()
-        val isClient = acceptor == null
-//        val name = if (isClient) {
-//            "CLIENT"
-//        } else {
-//            "SERVER"
-//        }
-//        channel.pipeline().addLast(DebugLoggingHandler("CONNECTION-$name"))
-
         val connection = NettyQuicConnection(
-            parentContext = channel.parent().attr(ATTRIBUTE_TRANSPORT_CONTEXT).get(),
+            parentContext = parentContext,
             channel = channel,
-            isClient = isClient
+            isClient = onConnection == null
         )
-        channel.attr(ATTRIBUTE_CONNECTION).set(connection)
+        channel.attr(NettyQuicConnection.ATTRIBUTE).set(connection)
 
-        //addLast(LoggingHandler(if (isClient) "CLIENT" else "SERVER"))
         channel.pipeline().addLast(
             "rsocket-connection",
             connection
         )
 
-        // initialize is run only for server
-        acceptor?.invoke(connection)
+        onConnection?.invoke(connection)
     }
 }