1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
// Copyright 2000-2021 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file.
@file:ApiStatus.Experimental
package com.intellij.openapi.progress
import com.intellij.openapi.util.Computable
import com.intellij.util.ConcurrencyUtil
import kotlinx.coroutines.*
import org.jetbrains.annotations.ApiStatus
import org.jetbrains.annotations.ApiStatus.Internal
import org.jetbrains.annotations.ApiStatus.ScheduledForRemoval
import kotlin.coroutines.coroutineContext
/**
* Checks whether the coroutine is active, and throws [CancellationException] if the coroutine is canceled.
* This function might suspend if the coroutine is paused,
* or yield if the coroutine has a lower priority while higher priority task is running.
*
* @throws CancellationException if the coroutine is canceled; the exception is also thrown if coroutine is canceled while suspended
* @see ensureActive
* @see coroutineSuspender
*/
suspend fun checkCanceled() {
val ctx = coroutineContext
ctx.ensureActive() // standard check first
ctx[CoroutineSuspenderElementKey]?.checkPaused() // will suspend if paused
}
/**
* The method has same semantics as [runBlocking], and additionally [action] gets canceled
* when [the current progress indicator][ProgressManager.getGlobalProgressIndicator] is cancelled,
* or [the current job][Cancellation.currentJob] is cancelled.
*
* This is a bridge for invoking suspending code from blocking code.
*
* Example:
* ```
* ProgressManager.getInstance().runProcess({
* runBlockingCancellable {
* someSuspendingFunctionWhichDoesntKnowAboutIndicator()
* }
* }, progress);
* ```
* @see runUnderIndicator
* @see runBlocking
*/
fun <T> runBlockingCancellable(action: suspend CoroutineScope.() -> T): T {
val indicator = ProgressManager.getGlobalProgressIndicator()
if (indicator != null) {
return runBlockingCancellable(indicator, action)
}
val currentJob = Cancellation.currentJob()
if (currentJob != null) {
// make runBlocking Job a child of the current one to propagate cancellation
return runBlocking(context = currentJob, block = action)
}
// we are not under indicator => just run the action, since nobody will cancel it anyway
return runBlocking(block = action)
}
fun <T> runBlockingCancellable(indicator: ProgressIndicator, action: suspend CoroutineScope.() -> T): T {
// we are under indicator => the Job must be canceled when indicator is canceled
return runBlocking(progressSinkElement(ProgressIndicatorSink(indicator)) + CoroutineName("indicator run blocking")) {
val indicatorWatchJob = launch(Dispatchers.IO + CoroutineName("indicator watcher")) {
while (true) {
if (indicator.isCanceled) {
// will throw PCE which will cancel the runBlocking Job and thrown further in the caller of runBlockingCancellable
indicator.checkCanceled()
}
delay(ConcurrencyUtil.DEFAULT_TIMEOUT_MS)
}
}
val result = action()
indicatorWatchJob.cancel()
result
}
}
/**
* Runs blocking (e.g. Java) code under indicator, which is canceled if current Job is canceled.
*
* This is a bridge for invoking blocking code from suspending code.
*
* Example:
* ```
* launch {
* runUnderIndicator {
* someJavaFunctionWhichDoesntKnowAboutCoroutines()
* }
* }
* ```
* @see runBlockingCancellable
* @see ProgressManager.runProcess
*/
suspend fun <T> runUnderIndicator(action: () -> T): T {
val ctx = coroutineContext
return runUnderIndicator(ctx.job, ctx.progressSink, action)
}
@OptIn(InternalCoroutinesApi::class)
fun <T> runUnderIndicator(job: Job, progressSink: ProgressSink?, action: () -> T): T {
job.ensureActive()
val indicator = if (progressSink == null) EmptyProgressIndicator() else ProgressSinkIndicator(progressSink)
try {
return ProgressManager.getInstance().runProcess(Computable {
// Register handler inside runProcess to avoid cancelling the indicator before even starting the progress.
// If the Job was canceled while runProcess was preparing,
// then CompletionHandler is invoked right away and cancels the indicator.
val completionHandle = job.invokeOnCompletion(onCancelling = true) {
if (it is CancellationException) {
indicator.cancel()
}
}
try {
indicator.checkCanceled()
action()
}
finally {
completionHandle.dispose()
}
}, indicator)
}
catch (e: ProcessCanceledException) {
if (!indicator.isCanceled) {
// means the exception was thrown manually
// => treat it as any other exception
throw e
}
// indicator is canceled
// => CompletionHandler was actually invoked
// => current Job is canceled
check(job.isCancelled)
throw job.getCancellationException()
}
}
fun CoroutineScope.progress(): Progress = JobProgress(coroutineContext.job)
@ScheduledForRemoval(inVersion = "2022.3")
@Deprecated(message = "Method was renamed", replaceWith = ReplaceWith("runBlockingCancellable(action)"))
fun <T> runSuspendingAction(action: suspend CoroutineScope.() -> T): T {
return runBlockingCancellable(action)
}
@ScheduledForRemoval(inVersion = "2022.3")
@Deprecated(message = "Method was renamed", replaceWith = ReplaceWith("runBlockingCancellable(indicator, action)"))
fun <T> runSuspendingAction(indicator: ProgressIndicator, action: suspend CoroutineScope.() -> T): T {
return runBlockingCancellable(indicator, action)
}
|
