minor netty tcp cleanup

Author: whyoleg

rsocket-transports/netty-tcp/src/jvmMain/kotlin/io/rsocket/kotlin/transport/netty/tcp/NettyTcpClientTransport.kt

@@ -91,7 +91,7 @@ private class NettyTcpClientTransportBuilderImpl : NettyTcpClientTransportBuilde
         }
 
         return NettyTcpClientTransportImpl(
-            coroutineContext = context.supervisorContext(),
+            coroutineContext = context.supervisorContext() + Dispatchers.Default,
             bootstrap = bootstrap,
             sslContext = sslContext,
         ).also {
@@ -121,8 +121,16 @@ private class NettyTcpClientTargetImpl(
     bootstrap: Bootstrap,
     sslContext: SslContext?,
     remoteAddress: SocketAddress,
-) : RSocketClientTarget, NettyTcpConnectionChannelInitializer(coroutineContext, sslContext) {
-    private val bootstrap = bootstrap.clone().handler(this).remoteAddress(remoteAddress)
+) : RSocketClientTarget {
+    private val bootstrap = bootstrap.clone()
+        .handler(
+            NettyTcpConnectionInitializer(
+                parentContext = coroutineContext,
+                sslContext = sslContext,
+                onConnection = null
+            )
+        )
+        .remoteAddress(remoteAddress)
 
     @RSocketTransportApi
     override suspend fun connectClient(): RSocketConnection {

rsocket-transports/netty-tcp/src/jvmMain/kotlin/io/rsocket/kotlin/transport/netty/tcp/NettyTcpConnection.kt

@@ -19,6 +19,8 @@ package io.rsocket.kotlin.transport.netty.tcp
 import io.netty.buffer.*
 import io.netty.channel.*
 import io.netty.channel.socket.*
+import io.netty.handler.codec.*
+import io.netty.handler.ssl.*
 import io.netty.util.*
 import io.rsocket.kotlin.internal.io.*
 import io.rsocket.kotlin.transport.*
@@ -44,7 +46,7 @@ internal class NettyTcpConnection(
     init {
         @OptIn(DelicateCoroutinesApi::class)
         launch(start = CoroutineStart.ATOMIC) {
-            val outboundJob = launch {
+            val outboundJob = launch(start = CoroutineStart.ATOMIC) {
                 nonCancellable {
                     try {
                         while (true) {
@@ -86,8 +88,8 @@ internal class NettyTcpConnection(
         cancel("exceptionCaught", cause)
     }
 
-    override fun userEventTriggered(ctx: ChannelHandlerContext?, evt: Any?) {
-        if (evt is ChannelInputShutdownEvent) inbound.close()
+    override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any?) {
+        if (evt === ChannelInputShutdownEvent.INSTANCE) inbound.close()
         super.userEventTriggered(ctx, evt)
     }
 
@@ -110,3 +112,43 @@ internal class NettyTcpConnection(
         val ATTRIBUTE: AttributeKey<RSocketConnection> = AttributeKey.newInstance<RSocketConnection>("rsocket-tcp-connection")
     }
 }
+
+@OptIn(RSocketTransportApi::class)
+internal class NettyTcpConnectionInitializer(
+    private val parentContext: CoroutineContext,
+    private val sslContext: SslContext?,
+    private val onConnection: ((RSocketConnection) -> Unit)?,
+) : ChannelInitializer<DuplexChannel>() {
+    override fun initChannel(channel: DuplexChannel) {
+        channel.config().isAutoRead = false
+
+        val connection = NettyTcpConnection(parentContext, channel)
+        channel.attr(NettyTcpConnection.ATTRIBUTE).set(connection)
+
+        if (sslContext != null) {
+            channel.pipeline().addLast("ssl", sslContext.newHandler(channel.alloc()))
+        }
+        channel.pipeline().addLast(
+            "rsocket-length-encoder",
+            LengthFieldPrepender(
+                /* lengthFieldLength = */ 3
+            )
+        )
+        channel.pipeline().addLast(
+            "rsocket-length-decoder",
+            LengthFieldBasedFrameDecoder(
+                /* maxFrameLength = */ Int.MAX_VALUE,
+                /* lengthFieldOffset = */ 0,
+                /* lengthFieldLength = */ 3,
+                /* lengthAdjustment = */ 0,
+                /* initialBytesToStrip = */ 3
+            )
+        )
+        channel.pipeline().addLast(
+            "rsocket-connection",
+            connection
+        )
+
+        onConnection?.invoke(connection)
+    }
+}

rsocket-transports/netty-tcp/src/jvmMain/kotlin/io/rsocket/kotlin/transport/netty/tcp/NettyTcpConnectionChannelInitializer.kt

@@ -20,7 +20,6 @@ import io.netty.bootstrap.*
 import io.netty.channel.*
 import io.netty.channel.ChannelFactory
 import io.netty.channel.nio.*
-import io.netty.channel.socket.*
 import io.netty.channel.socket.nio.*
 import io.netty.handler.ssl.*
 import io.rsocket.kotlin.internal.io.*
@@ -111,7 +110,7 @@ private class NettyTcpServerTransportBuilderImpl : NettyTcpServerTransportBuilde
         }
 
         return NettyTcpServerTransportImpl(
-            coroutineContext = context.supervisorContext(),
+            coroutineContext = context.supervisorContext() + Dispatchers.Default,
             bootstrap = bootstrap,
             sslContext = sslContext,
         ).also {
@@ -149,8 +148,8 @@ private class NettyTcpServerTargetImpl(
 
         val instanceContext = coroutineContext.childContext()
         val channel = try {
-            val handler = NettyTcpConnectionServerInitializer(
-                coroutineContext = instanceContext.supervisorContext(),
+            val handler = NettyTcpConnectionInitializer(
+                parentContext = instanceContext.supervisorContext(),
                 sslContext = sslContext,
                 onConnection = onConnection,
             )
@@ -187,15 +186,3 @@ private class NettyTcpServerInstanceImpl(
         }
     }
 }
-
-@RSocketTransportApi
-private class NettyTcpConnectionServerInitializer(
-    coroutineContext: CoroutineContext,
-    sslContext: SslContext?,
-    private val onConnection: (RSocketConnection) -> Unit,
-) : NettyTcpConnectionChannelInitializer(coroutineContext, sslContext) {
-    override fun initChannel(ch: DuplexChannel) {
-        super.initChannel(ch)
-        onConnection(ch.attr(NettyTcpConnection.ATTRIBUTE).get())
-    }
-}