ensures Payload released in LimitingFlowCollector (#172)

Author: OlegDokuka

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/flow/LimitingFlowCollector.kt

@@ -39,7 +39,12 @@ internal class LimitingFlowCollector(
     }
 
     override suspend fun emit(value: Payload): Unit = value.closeOnError {
-        useRequest()
+        try {
+            useRequest()
+        } catch (t: Throwable) {
+            value.release()
+            throw t
+        }
         state.send(NextPayloadFrame(streamId, value))
     }
 

rsocket-transport-ktor/src/commonMain/kotlin/io/rsocket/kotlin/transport/ktor/TcpConnection.kt

@@ -42,6 +42,7 @@ internal class TcpConnection(private val socket: Socket) : Connection, Coroutine
     private val receiveChannel = SafeChannel<ByteReadPacket>(8)
 
     init {
+        val channelCloseJob = Job(job)
         launch {
             socket.openWriteChannel(autoFlush = true).use {
                 while (isActive) {
@@ -77,6 +78,14 @@ internal class TcpConnection(private val socket: Socket) : Connection, Coroutine
             val error = cause?.let { it as? CancellationException ?: CancellationException("Connection failed", it) }
             sendChannel.cancel(error)
             receiveChannel.cancel(error)
+            CoroutineScope(job).launch {
+                while (!sendChannel.isClosedForReceive || !sendChannel.isClosedForSend
+                    || !receiveChannel.isClosedForReceive || !receiveChannel.isClosedForSend
+                ) {
+                    delay(1)
+                }
+                channelCloseJob.complete()
+            }
         }
     }
 

rsocket-transport-local/src/commonMain/kotlin/io/rsocket/kotlin/transport/local/LocalServer.kt

@@ -44,10 +44,19 @@ internal constructor(
         val clientChannel = SafeChannel<ByteReadPacket>(Channel.UNLIMITED)
         val serverChannel = SafeChannel<ByteReadPacket>(Channel.UNLIMITED)
         val connectionJob = Job(job)
+        val channelCloseJob = Job(job)
         connectionJob.invokeOnCompletion {
             val error = CancellationException("Connection failed", it)
             clientChannel.cancel(error)
             serverChannel.cancel(error)
+            CoroutineScope(job).launch {
+                while (!clientChannel.isClosedForReceive || !clientChannel.isClosedForSend
+                    || !serverChannel.isClosedForReceive || !serverChannel.isClosedForSend
+                ) {
+                    delay(1)
+                }
+                channelCloseJob.complete()
+            }
         }
         val clientConnection = LocalConnection(serverChannel, clientChannel, pool, connectionJob)
         val serverConnection = LocalConnection(clientChannel, serverChannel, pool, connectionJob)