fix some issues

rsocket · Mar 12, 2024 · eb4677e · eb4677e
1 parent 5e5620b
commit eb4677e
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequestFlow.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequestFlow.kt
@@ -41,7 +41,13 @@ internal suspend fun FlowCollector<Payload>.emitAllWithRequestN(
     while (true) {
         val result = payloads.receiveCatching()
         if (result.isClosed) return result.exceptionOrNull()
-        emit(result.getOrThrow()) // will never throw
+        val payload = result.getOrThrow() // will never throw
+        try {
+            emit(payload)
+        } catch (cause: Throwable) {
+            payload.close()
+            throw cause
+        }
 
         @OptIn(DelicateCoroutinesApi::class)
         if (requestNs.isClosedForSend) continue

diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/TestConnection.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/TestConnection.kt
@@ -39,7 +39,7 @@ class TestConnection : RSocketTransportSession.Sequential, RSocketClientTarget {
 
     init {
         coroutineContext.job.invokeOnCompletion {
-            sendChannel.close(it)
+            sendChannel.cancelWithCause(it)
             receiveChannel.cancelWithCause(it)
         }
     }

diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt
@@ -51,8 +51,12 @@ class RSocketTest : SuspendTest, TestWithLeakCheck {
                 }
                 requestChannel { init, payloads ->
                     init.close()
-                    payloads.onEach { it.close() }.launchIn(this)
-                    flow { repeat(10) { emitOrClose(payload("server got -> [$it]")) } }
+                    flow {
+                        coroutineScope {
+                            payloads.onEach { it.close() }.launchIn(this)
+                            repeat(10) { emitOrClose(payload("server got -> [$it]")) }
+                        }
+                    }
                 }
             }
         }