Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 1 | /* |
Roman Elizarov | 1f74a2d | 2018-06-29 19:19:45 +0300 | [diff] [blame^] | 2 | * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. |
Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 5 | package kotlinx.coroutines.experimental.future |
| 6 | |
| 7 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | a4b5693 | 2018-03-13 17:59:48 +0300 | [diff] [blame] | 8 | import java.util.concurrent.* |
| 9 | import java.util.function.* |
| 10 | import kotlin.coroutines.experimental.* |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 11 | |
| 12 | /** |
Roman Elizarov | b170819 | 2017-12-22 12:14:05 +0300 | [diff] [blame] | 13 | * Starts new coroutine and returns its result as an implementation of [CompletableFuture]. |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 14 | * This coroutine builder uses [CommonPool] context by default and is conceptually similar to [CompletableFuture.supplyAsync]. |
| 15 | * |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 16 | * The running coroutine is cancelled when the resulting future is cancelled or otherwise completed. |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 17 | * |
| 18 | * The [context] for the new coroutine can be explicitly specified. |
| 19 | * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`. |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 20 | * The [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html) |
| 21 | * of the parent coroutine may be used, |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 22 | * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine. |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 23 | * The parent job may be also explicitly specified using [parent] parameter. |
| 24 | * |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 25 | * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used. |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 26 | * |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 27 | * By default, the coroutine is immediately scheduled for execution. |
| 28 | * Other options can be specified via `start` parameter. See [CoroutineStart] for details. |
| 29 | * A value of [CoroutineStart.LAZY] is not supported |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 30 | * (since `CompletableFuture` framework does not provide the corresponding capability) and |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 31 | * produces [IllegalArgumentException]. |
| 32 | * |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 33 | * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine. |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 34 | * |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 35 | * @param context context of the coroutine. The default value is [DefaultDispatcher]. |
| 36 | * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT]. |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 37 | * @param parent explicitly specifies the parent job, overrides job from the [context] (if any). |
Roman Elizarov | a4b5693 | 2018-03-13 17:59:48 +0300 | [diff] [blame] | 38 | * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]). |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 39 | * @param block the coroutine code. |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 40 | */ |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 41 | public fun <T> future( |
Roman Elizarov | c0d559b | 2017-09-28 14:27:05 +0300 | [diff] [blame] | 42 | context: CoroutineContext = DefaultDispatcher, |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 43 | start: CoroutineStart = CoroutineStart.DEFAULT, |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 44 | parent: Job? = null, |
Roman Elizarov | a4b5693 | 2018-03-13 17:59:48 +0300 | [diff] [blame] | 45 | onCompletion: CompletionHandler? = null, |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 46 | block: suspend CoroutineScope.() -> T |
| 47 | ): CompletableFuture<T> { |
| 48 | require(!start.isLazy) { "$start start is not supported" } |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 49 | val newContext = newCoroutineContext(context, parent) |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 50 | val job = Job(newContext[Job]) |
| 51 | val future = CompletableFutureCoroutine<T>(newContext + job) |
| 52 | job.cancelFutureOnCompletion(future) |
| 53 | future.whenComplete { _, exception -> job.cancel(exception) } |
Roman Elizarov | a4b5693 | 2018-03-13 17:59:48 +0300 | [diff] [blame] | 54 | if (onCompletion != null) job.invokeOnCompletion(handler = onCompletion) |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 55 | start(block, receiver=future, completion=future) // use the specified start strategy |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 56 | return future |
| 57 | } |
| 58 | |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 59 | /** @suppress **Deprecated**: Binary compatibility */ |
| 60 | @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN) |
| 61 | public fun <T> future( |
| 62 | context: CoroutineContext = DefaultDispatcher, |
| 63 | start: CoroutineStart = CoroutineStart.DEFAULT, |
Roman Elizarov | a4b5693 | 2018-03-13 17:59:48 +0300 | [diff] [blame] | 64 | parent: Job? = null, |
| 65 | block: suspend CoroutineScope.() -> T |
| 66 | ): CompletableFuture<T> = future(context, start, parent, block = block) |
| 67 | |
| 68 | /** @suppress **Deprecated**: Binary compatibility */ |
| 69 | @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN) |
| 70 | public fun <T> future( |
| 71 | context: CoroutineContext = DefaultDispatcher, |
| 72 | start: CoroutineStart = CoroutineStart.DEFAULT, |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 73 | block: suspend CoroutineScope.() -> T |
| 74 | ): CompletableFuture<T> = |
| 75 | future(context, start, block = block) |
| 76 | |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 77 | private class CompletableFutureCoroutine<T>( |
| 78 | override val context: CoroutineContext |
| 79 | ) : CompletableFuture<T>(), Continuation<T>, CoroutineScope { |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 80 | override val coroutineContext: CoroutineContext get() = context |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 81 | override val isActive: Boolean get() = context[Job]!!.isActive |
| 82 | override fun resume(value: T) { complete(value) } |
| 83 | override fun resumeWithException(exception: Throwable) { completeExceptionally(exception) } |
| 84 | } |
| 85 | |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 86 | /** |
| 87 | * Converts this deferred value to the instance of [CompletableFuture]. |
| 88 | * The deferred value is cancelled when the resulting future is cancelled or otherwise completed. |
| 89 | */ |
Roman Elizarov | f7a7f7b | 2017-03-09 12:08:36 +0300 | [diff] [blame] | 90 | public fun <T> Deferred<T>.asCompletableFuture(): CompletableFuture<T> { |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 91 | val future = CompletableFuture<T>() |
| 92 | future.whenComplete { _, exception -> cancel(exception) } |
Roman Elizarov | e780347 | 2017-02-16 09:52:31 +0300 | [diff] [blame] | 93 | invokeOnCompletion { |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 94 | try { |
Roman Elizarov | c581454 | 2017-01-19 10:19:06 +0300 | [diff] [blame] | 95 | future.complete(getCompleted()) |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 96 | } catch (exception: Exception) { |
| 97 | future.completeExceptionally(exception) |
| 98 | } |
| 99 | } |
| 100 | return future |
| 101 | } |
| 102 | |
| 103 | /** |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 104 | * Converts this completion stage to an instance of [Deferred]. |
| 105 | * When this completion stage is an instance of [Future], then it is cancelled when |
| 106 | * the resulting deferred is cancelled. |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 107 | */ |
| 108 | public fun <T> CompletionStage<T>.asDeferred(): Deferred<T> { |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 109 | // Fast path if already completed |
| 110 | if (this is Future<*> && isDone()){ |
| 111 | return try { |
| 112 | @Suppress("UNCHECKED_CAST") |
| 113 | CompletableDeferred(get() as T) |
Roman Elizarov | 19c1f2e | 2018-03-01 22:15:58 +0300 | [diff] [blame] | 114 | } catch (e: Throwable) { |
| 115 | // unwrap original cause from ExecutionException |
| 116 | val original = (e as? ExecutionException)?.cause ?: e |
| 117 | CompletableDeferred<T>().also { it.completeExceptionally(original) } |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 118 | } |
| 119 | } |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 120 | val result = CompletableDeferred<T>() |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 121 | whenComplete { value, exception -> |
| 122 | if (exception == null) { |
| 123 | result.complete(value) |
| 124 | } else { |
| 125 | result.completeExceptionally(exception) |
| 126 | } |
| 127 | } |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 128 | if (this is Future<*>) result.cancelFutureOnCompletion(this) |
Jonathan Cornaz | 7bd2c50 | 2018-02-28 15:59:21 +0100 | [diff] [blame] | 129 | return result |
| 130 | } |
| 131 | |
| 132 | /** |
Roman Elizarov | be4cae3 | 2017-02-15 17:57:02 +0300 | [diff] [blame] | 133 | * Awaits for completion of the future without blocking a thread. |
| 134 | * |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 135 | * @suppress **Deprecated**: For binary compatibility only |
| 136 | */ |
| 137 | @Deprecated("For binary compatibility only", level = DeprecationLevel.HIDDEN) |
| 138 | public suspend fun <T> CompletableFuture<T>.await(): T = |
| 139 | (this as CompletionStage<T>).await() |
| 140 | |
| 141 | /** |
| 142 | * Awaits for completion of the completion stage without blocking a thread. |
| 143 | * |
Roman Elizarov | be4cae3 | 2017-02-15 17:57:02 +0300 | [diff] [blame] | 144 | * This suspending function is cancellable. |
Roman Elizarov | d82b3a9 | 2017-06-23 21:52:08 +0300 | [diff] [blame] | 145 | * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 146 | * stops waiting for the completion stage and immediately resumes with [CancellationException]. |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 147 | * |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 148 | * Note, that `CompletionStage` implementation does not support prompt removal of installed listeners, so on cancellation of this wait |
| 149 | * a few small objects will remain in the `CompletionStage` stack of completion actions until it completes itself. |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 150 | * However, the care is taken to clear the reference to the waiting coroutine itself, so that its memory can be |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 151 | * released even if the completion stage never completes. |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 152 | */ |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 153 | public suspend fun <T> CompletionStage<T>.await(): T { |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 154 | // fast path when CompletableFuture is already done (does not suspend) |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 155 | if (this is Future<*> && isDone()) { |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 156 | try { |
Vsevolod Tolstopyatov | 87f2faa | 2018-04-30 22:53:02 +0300 | [diff] [blame] | 157 | @Suppress("UNCHECKED_CAST") |
Roman Elizarov | eb4f9be | 2018-03-01 22:41:08 +0300 | [diff] [blame] | 158 | return get() as T |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 159 | } catch (e: ExecutionException) { |
| 160 | throw e.cause ?: e // unwrap original cause from ExecutionException |
Roman Elizarov | ee89344 | 2017-01-19 14:56:21 +0300 | [diff] [blame] | 161 | } |
Roman Elizarov | ee89344 | 2017-01-19 14:56:21 +0300 | [diff] [blame] | 162 | } |
Roman Elizarov | fc6461f | 2017-05-16 18:48:17 +0300 | [diff] [blame] | 163 | // slow path -- suspend |
Roman Elizarov | ee89344 | 2017-01-19 14:56:21 +0300 | [diff] [blame] | 164 | return suspendCancellableCoroutine { cont: CancellableContinuation<T> -> |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 165 | val consumer = ContinuationConsumer(cont) |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 166 | whenComplete(consumer) |
Vsevolod Tolstopyatov | 80a2947 | 2018-04-17 16:02:02 +0300 | [diff] [blame] | 167 | cont.invokeOnCancellation { |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 168 | consumer.cont = null // shall clear reference to continuation |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 169 | } |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 170 | } |
Roman Elizarov | ee89344 | 2017-01-19 14:56:21 +0300 | [diff] [blame] | 171 | } |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 172 | |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 173 | private class ContinuationConsumer<T>( |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 174 | @Volatile @JvmField var cont: Continuation<T>? |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 175 | ) : BiConsumer<T?, Throwable?> { |
Roman Elizarov | 63e779c | 2017-07-11 14:42:48 +0300 | [diff] [blame] | 176 | @Suppress("UNCHECKED_CAST") |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 177 | override fun accept(result: T?, exception: Throwable?) { |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 178 | val cont = this.cont ?: return // atomically read current value unless null |
Vsevolod Tolstopyatov | 1cbe8f0 | 2018-06-05 18:13:51 +0300 | [diff] [blame] | 179 | if (exception == null) { |
| 180 | // the future has been completed normally |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 181 | cont.resume(result as T) |
Vsevolod Tolstopyatov | 1cbe8f0 | 2018-06-05 18:13:51 +0300 | [diff] [blame] | 182 | } else { |
| 183 | // the future has completed with an exception, unwrap it to provide consistent view of .await() result and to propagate only original exception |
| 184 | cont.resumeWithException((exception as? CompletionException)?.cause ?: exception) |
| 185 | } |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 186 | } |
Roman Elizarov | 3754f95 | 2017-01-18 20:47:54 +0300 | [diff] [blame] | 187 | } |
Roman Elizarov | fc6461f | 2017-05-16 18:48:17 +0300 | [diff] [blame] | 188 | |
| 189 | // --------------------------------------- DEPRECATED APIs --------------------------------------- |
| 190 | // We keep it only for backwards compatibility with old versions of this integration library. |
| 191 | // Do not copy when using this file an example for other integration. |
| 192 | |
| 193 | /** |
| 194 | * Converts this deferred value to the instance of [CompletableFuture]. |
| 195 | * The deferred value is cancelled when the resulting future is cancelled or otherwise completed. |
| 196 | * @suppress: **Deprecated**: Renamed to [asCompletableFuture] |
| 197 | */ |
| 198 | @Deprecated("Renamed to `asCompletableFuture`", |
| 199 | replaceWith = ReplaceWith("asCompletableFuture()")) |
| 200 | public fun <T> Deferred<T>.toCompletableFuture(): CompletableFuture<T> = asCompletableFuture() |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 201 | |
Roman Elizarov | 4eae2a8 | 2017-05-17 20:55:27 +0300 | [diff] [blame] | 202 | /** @suppress **Deprecated** */ |
Roman Elizarov | 489cac2 | 2017-05-17 12:18:28 +0300 | [diff] [blame] | 203 | @Suppress("DeprecatedCallableAddReplaceWith") // todo: the warning is incorrectly shown, see KT-17917 |
Roman Elizarov | 3fa4bec | 2017-05-16 21:05:43 +0300 | [diff] [blame] | 204 | @Deprecated("Use the other version. This one is for binary compatibility only.", level=DeprecationLevel.HIDDEN) |
| 205 | public fun <T> future( |
| 206 | context: CoroutineContext = CommonPool, |
| 207 | block: suspend () -> T |
Vsevolod Tolstopyatov | 1cbe8f0 | 2018-06-05 18:13:51 +0300 | [diff] [blame] | 208 | ): CompletableFuture<T> = future(context = context) { block() } |