fix leaks

rsocket · Mar 12, 2024 · 238909a · 238909a
1 parent 5e1faf7
commit 238909a
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 4 deletions.
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequesterOperation.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/RequesterOperation.kt
@@ -49,13 +49,20 @@ internal class RequesterOperationExecutor(
             requestsScope.ensureActive()
         }
 
-        return requestsScope.launch {
+        @OptIn(ExperimentalCoroutinesApi::class)
+        return requestsScope.launch(start = CoroutineStart.ATOMIC) {
+            try {
+                ensureActive()
+            } catch (cause: Throwable) {
+                payload.close()
+                operation.receiveProcessingError(cause)
+                throw cause
+            }
             val outbound = try {
                 operationFactory.createRequest(operation.type, OperationFrameHandler(operation))
             } catch (cause: Throwable) {
-                operation.receiveProcessingError(cause)
                 payload.close()
-
+                operation.receiveProcessingError(cause)
                 throw cause
             }
 

diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/ResponderOperation.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/operation/ResponderOperation.kt
@@ -76,7 +76,14 @@ private class WrappedResponderOperation(
             null -> {
                 // TODO: need to somehow enforce this better
                 checkNotNull(payload) { "Payload should be present for request" }
-                requestJob = requestsScope.launch {
+                @OptIn(ExperimentalCoroutinesApi::class)
+                requestJob = requestsScope.launch(start = CoroutineStart.ATOMIC) {
+                    try {
+                        ensureActive()
+                    } catch (cause: Throwable) {
+                        payload.close()
+                        throw cause
+                    }
                     try {
                         operation.execute(outbound, payload, complete)
                     } catch (cause: Throwable) {