fix netty QUIC

Author: whyoleg

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

@@ -48,7 +48,11 @@ public fun Buffer.toByteBuf(allocator: ByteBufAllocator): ByteBuf {
     return nettyBuffer
 }
 
-// TODO: handle FUTURE result at least in QUIC
 public fun Channel.writeBuffer(buffer: Buffer) {
     write(buffer.toByteBuf(alloc()), voidPromise())
 }
+
+// for quic
+public suspend fun Channel.writeAndFlushBuffer(buffer: Buffer) {
+    writeAndFlush(buffer.toByteBuf(alloc())).awaitFuture()
+}

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

@@ -28,7 +28,7 @@ internal val QuicStreamChannel.debugName: String
         } else {
             "REMOTE"
         }
-        return "STREAM-$name"
+        return "STREAM-$name $this"
     }
 
 // TODO: move it somehow to tests

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

@@ -25,7 +25,6 @@ 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.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.io.*
 import kotlin.coroutines.*
@@ -47,33 +46,24 @@ internal class NettyQuicStream(
     init {
         @OptIn(DelicateCoroutinesApi::class)
         launch(start = CoroutineStart.ATOMIC) {
-            val outboundJob = launch {
+            launch {
+                // TODO: check cancellation?
                 nonCancellable {
                     try {
                         while (true) {
-                            // we write all available frames here, and only after it flush
-                            // in this case, if there are several buffered frames we can send them in one go
-                            // avoiding unnecessary flushes
-                            channel.writeBuffer(outbound.receiveCatching().getOrNull() ?: break)
-                            while (true) channel.writeBuffer(outbound.tryReceive().getOrNull() ?: break)
-                            channel.flush()
+                            channel.writeAndFlushBuffer(outbound.receiveCatching().getOrNull() ?: break)
                         }
                     } finally {
                         outbound.cancel()
-                        // we dont' use `shutdownOutput` here as it will not flush buffered messages!
-                        channel.writeAndFlush(QuicStreamFrame.EMPTY_FIN).awaitFuture()
+                        channel.shutdownOutput().awaitFuture()
                     }
                 }
             }
             try {
                 awaitCancellation()
             } finally {
-                nonCancellable {
-                    outbound.close()
-                    inbound.cancel()
-                    outboundJob.join()
-                    channel.close().awaitFuture()
-                }
+                outbound.close()
+                inbound.cancel()
             }
         }
     }
@@ -87,8 +77,8 @@ internal class NettyQuicStream(
         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)
     }
 
@@ -106,8 +96,6 @@ internal class NettyQuicStream(
     }
 
     override suspend fun receiveFrame(): Buffer? {
-        inbound.tryReceive().onSuccess { return it }
-        channel.read()
         return inbound.receiveCatching().getOrNull()
     }
 }
@@ -116,7 +104,6 @@ internal class NettyQuicStream(
 internal object NettyQuicStreamInitializer : ChannelInitializer<QuicStreamChannel>() {
     override fun initChannel(channel: QuicStreamChannel) {
 //        channel.pipeline().addLast(DebugLoggingHandler(channel.debugName))
-        channel.config().isAutoRead = false
 
         channel.pipeline().addLast(
             "rsocket-length-encoder",