blob: 8a1b7bdffde5e5bbaa8c9c6e3b88d4eb052fdbf6 (
plain)
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
150
151
152
153
154
155
156
157
158
159
160
161
162
|
@file:Suppress("RedundantVisibilityModifier")
package kotlinx.coroutines.tasks
import com.google.android.gms.tasks.*
import kotlinx.coroutines.*
import java.lang.Runnable
import java.util.concurrent.Executor
import kotlin.coroutines.*
/**
* Converts this deferred to the instance of [Task].
* If deferred is cancelled then resulting task will be cancelled as well.
*/
public fun <T> Deferred<T>.asTask(): Task<T> {
val cancellation = CancellationTokenSource()
val source = TaskCompletionSource<T>(cancellation.token)
invokeOnCompletion callback@{
if (it is CancellationException) {
cancellation.cancel()
return@callback
}
val t = getCompletionExceptionOrNull()
if (t == null) {
source.setResult(getCompleted())
} else {
source.setException(t as? Exception ?: RuntimeExecutionException(t))
}
}
return source.task
}
/**
* Converts this task to an instance of [Deferred].
* If task is cancelled then resulting deferred will be cancelled as well.
* However, the opposite is not true: if the deferred is cancelled, the [Task] will not be cancelled.
* For bi-directional cancellation, an overload that accepts [CancellationTokenSource] can be used.
*/
public fun <T> Task<T>.asDeferred(): Deferred<T> = asDeferredImpl(null)
/**
* Converts this task to an instance of [Deferred] with a [CancellationTokenSource] to control cancellation.
* The cancellation of this function is bi-directional:
* * If the given task is cancelled, the resulting deferred will be cancelled.
* * If the resulting deferred is cancelled, the provided [cancellationTokenSource] will be cancelled.
*
* Providing a [CancellationTokenSource] that is unrelated to the receiving [Task] is not supported and
* leads to an unspecified behaviour.
*/
@ExperimentalCoroutinesApi // Since 1.5.1, tentatively until 1.6.0
public fun <T> Task<T>.asDeferred(cancellationTokenSource: CancellationTokenSource): Deferred<T> =
asDeferredImpl(cancellationTokenSource)
private fun <T> Task<T>.asDeferredImpl(cancellationTokenSource: CancellationTokenSource?): Deferred<T> {
val deferred = CompletableDeferred<T>()
if (isComplete) {
val e = exception
if (e == null) {
if (isCanceled) {
deferred.cancel()
} else {
@Suppress("UNCHECKED_CAST")
deferred.complete(result as T)
}
} else {
deferred.completeExceptionally(e)
}
} else {
// Run the callback directly to avoid unnecessarily scheduling on the main thread.
addOnCompleteListener(DirectExecutor) {
val e = it.exception
if (e == null) {
@Suppress("UNCHECKED_CAST")
if (it.isCanceled) deferred.cancel() else deferred.complete(it.result as T)
} else {
deferred.completeExceptionally(e)
}
}
}
if (cancellationTokenSource != null) {
deferred.invokeOnCompletion {
cancellationTokenSource.cancel()
}
}
// Prevent casting to CompletableDeferred and manual completion.
return object : Deferred<T> by deferred {}
}
/**
* Awaits the completion of the task without blocking a thread.
*
* This suspending function is cancellable.
* If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
* stops waiting for the completion stage and immediately resumes with [CancellationException].
*
* For bi-directional cancellation, an overload that accepts [CancellationTokenSource] can be used.
*/
public suspend fun <T> Task<T>.await(): T = awaitImpl(null)
/**
* Awaits the completion of the task that is linked to the given [CancellationTokenSource] to control cancellation.
*
* This suspending function is cancellable and cancellation is bi-directional:
* * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
* cancels the [cancellationTokenSource] and throws a [CancellationException].
* * If the task is cancelled, then this function will throw a [CancellationException].
*
* Providing a [CancellationTokenSource] that is unrelated to the receiving [Task] is not supported and
* leads to an unspecified behaviour.
*/
@ExperimentalCoroutinesApi // Since 1.5.1, tentatively until 1.6.0
public suspend fun <T> Task<T>.await(cancellationTokenSource: CancellationTokenSource): T =
awaitImpl(cancellationTokenSource)
private suspend fun <T> Task<T>.awaitImpl(cancellationTokenSource: CancellationTokenSource?): T {
// fast path
if (isComplete) {
val e = exception
return if (e == null) {
if (isCanceled) {
throw CancellationException("Task $this was cancelled normally.")
} else {
@Suppress("UNCHECKED_CAST")
result as T
}
} else {
throw e
}
}
return suspendCancellableCoroutine { cont ->
// Run the callback directly to avoid unnecessarily scheduling on the main thread.
addOnCompleteListener(DirectExecutor) {
val e = it.exception
if (e == null) {
@Suppress("UNCHECKED_CAST")
if (it.isCanceled) cont.cancel() else cont.resume(it.result as T)
} else {
cont.resumeWithException(e)
}
}
if (cancellationTokenSource != null) {
cont.invokeOnCancellation {
cancellationTokenSource.cancel()
}
}
}
}
/**
* An [Executor] that just directly executes the [Runnable].
*/
private object DirectExecutor : Executor {
override fun execute(r: Runnable) {
r.run()
}
}
|
