diff --git a/platform/xdebugger-impl/frontend/src/com/intellij/platform/debugger/impl/frontend/util/SequentialRpcRequestsExecutor.kt b/platform/xdebugger-impl/frontend/src/com/intellij/platform/debugger/impl/frontend/util/SequentialRpcRequestsExecutor.kt index 763a98103776..516540dcfcd8 100644 --- a/platform/xdebugger-impl/frontend/src/com/intellij/platform/debugger/impl/frontend/util/SequentialRpcRequestsExecutor.kt +++ b/platform/xdebugger-impl/frontend/src/com/intellij/platform/debugger/impl/frontend/util/SequentialRpcRequestsExecutor.kt @@ -9,10 +9,8 @@ import kotlinx.coroutines.Deferred import kotlinx.coroutines.async import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.consumeEach -import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.completeWith import kotlinx.coroutines.launch -import kotlinx.coroutines.selects.select import kotlinx.coroutines.supervisorScope /** @@ -58,22 +56,21 @@ internal class SequentialRpcRequestsExecutor private constructor() { override suspend fun performRequest() { if (!result.isActive) return - - supervisorScope { - val requestResult = async { request() } - select { - requestResult.onJoin { - result.completeWith(runCatching { requestResult.await() }) - } - result.onJoin { - requestResult.cancelAndJoin() - } + try { + supervisorScope { + val requestResult = async { request() } + result.invokeOnCompletion { requestResult.cancel() } + result.completeWith(runCatching { requestResult.await() }) } } + finally { + result.cancel() + } + } override fun markUndelivered() { - this.result.cancel() + result.cancel() } } diff --git a/platform/xdebugger-impl/frontend/tests/com/intellij/platform/debugger/impl/frontend/tests/SequentialRpcRequestsExecutorTest.kt b/platform/xdebugger-impl/frontend/tests/com/intellij/platform/debugger/impl/frontend/tests/SequentialRpcRequestsExecutorTest.kt index b2f96ce2d5fe..9850e3a582b2 100644 --- a/platform/xdebugger-impl/frontend/tests/com/intellij/platform/debugger/impl/frontend/tests/SequentialRpcRequestsExecutorTest.kt +++ b/platform/xdebugger-impl/frontend/tests/com/intellij/platform/debugger/impl/frontend/tests/SequentialRpcRequestsExecutorTest.kt @@ -5,6 +5,7 @@ import com.intellij.platform.debugger.impl.frontend.util.SequentialRpcRequestsEx import com.intellij.platform.util.coroutines.childScope import com.intellij.testFramework.LoggedErrorProcessor import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.cancel import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout @@ -155,6 +156,28 @@ internal class SequentialRpcRequestsExecutorTest { assertTrue(request.isCancelled) } + @Test + fun `cancelling executor scope cancels running and queued requests`() = runBlocking { + val scope = childScope("SequentialRpcRequestsExecutor") + val executor = SequentialRpcRequestsExecutor.create(scope) + val requestStarted = CompletableDeferred() + val runningRequest = executor.submit { + requestStarted.complete(Unit) + awaitCancellation() + } + val queuedRequest = executor.submit { + error("Queued request must not be executed") + } + + requestStarted.await() + scope.cancel() + runningRequest.join() + queuedRequest.join() + + assertTrue(runningRequest.isCancelled) + assertTrue(queuedRequest.isCancelled) + } + private fun runTest(test: suspend (SequentialRpcRequestsExecutor) -> Unit) = runBlocking { val scope = childScope("SequentialRpcRequestsExecutor") val executor = SequentialRpcRequestsExecutor.create(scope)