1 /*
<lambda>null2 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
3 */
4
5 @file:Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")
6
7 package kotlinx.coroutines.reactor
8
9 import kotlinx.coroutines.*
10 import kotlinx.coroutines.reactive.*
11 import org.reactivestreams.*
12 import reactor.core.*
13 import reactor.core.publisher.*
14 import kotlin.coroutines.*
15 import kotlinx.coroutines.internal.*
16
17 /**
18 * Creates a cold [mono][Mono] that runs a given [block] in a coroutine and emits its result.
19 * Every time the returned mono is subscribed, it starts a new coroutine.
20 * If the result of [block] is `null`, [MonoSink.success] is invoked without a value.
21 * Unsubscribing cancels the running coroutine.
22 *
23 * Coroutine context can be specified with [context] argument.
24 * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
25 *
26 * @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
27 */
28 public fun <T> mono(
29 context: CoroutineContext = EmptyCoroutineContext,
30 block: suspend CoroutineScope.() -> T?
31 ): Mono<T> {
32 require(context[Job] === null) { "Mono context cannot contain job in it." +
33 "Its lifecycle should be managed via Disposable handle. Had $context" }
34 return monoInternal(GlobalScope, context, block)
35 }
36
37 /**
38 * Awaits the single value from the given [Mono] without blocking the thread and returns the resulting value, or, if
39 * this publisher has produced an error, throws the corresponding exception. If the Mono completed without a value,
40 * `null` is returned.
41 *
42 * This suspending function is cancellable.
43 * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
44 * function immediately cancels its [Subscription] and resumes with [CancellationException].
45 */
awaitSingleOrNullnull46 public suspend fun <T> Mono<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
47 injectCoroutineContext(cont.context).subscribe(object : Subscriber<T> {
48 private var seenValue = false
49
50 override fun onSubscribe(s: Subscription) {
51 cont.invokeOnCancellation { s.cancel() }
52 s.request(Long.MAX_VALUE)
53 }
54
55 override fun onComplete() {
56 if (!seenValue) cont.resume(null)
57 }
58
59 override fun onNext(t: T) {
60 seenValue = true
61 cont.resume(t)
62 }
63
64 override fun onError(error: Throwable) { cont.resumeWithException(error) }
65 })
66 }
67
68 /**
69 * Awaits the single value from the given [Mono] without blocking the thread and returns the resulting value, or,
70 * if this Mono has produced an error, throws the corresponding exception.
71 *
72 * This suspending function is cancellable.
73 * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
74 * function immediately cancels its [Subscription] and resumes with [CancellationException].
75 *
76 * @throws NoSuchElementException if the Mono does not emit any value
77 */
78 // TODO: consider using https://github.com/Kotlin/kotlinx.coroutines/issues/2607 once that lands
awaitSinglenull79 public suspend fun <T> Mono<T>.awaitSingle(): T = awaitSingleOrNull() ?: throw NoSuchElementException()
80
81 private fun <T> monoInternal(
82 scope: CoroutineScope, // support for legacy mono in scope
83 context: CoroutineContext,
84 block: suspend CoroutineScope.() -> T?
85 ): Mono<T> = Mono.create { sink ->
86 val reactorContext = context.extendReactorContext(sink.currentContext())
87 val newContext = scope.newCoroutineContext(context + reactorContext)
88 val coroutine = MonoCoroutine(newContext, sink)
89 sink.onDispose(coroutine)
90 coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
91 }
92
93 private class MonoCoroutine<in T>(
94 parentContext: CoroutineContext,
95 private val sink: MonoSink<T>
96 ) : AbstractCoroutine<T>(parentContext, false, true), Disposable {
97 @Volatile
98 private var disposed = false
99
onCompletednull100 override fun onCompleted(value: T) {
101 if (value == null) sink.success() else sink.success(value)
102 }
103
onCancellednull104 override fun onCancelled(cause: Throwable, handled: Boolean) {
105 /** Cancellation exceptions that were caused by [dispose], that is, came from downstream, are not errors. */
106 val unwrappedCause = unwrap(cause)
107 if (getCancellationException() !== unwrappedCause || !disposed) {
108 try {
109 /** If [sink] turns out to already be in a terminal state, this exception will be passed through the
110 * [Hooks.onOperatorError] hook, which is the way to signal undeliverable exceptions in Reactor. */
111 sink.error(cause)
112 } catch (e: Throwable) {
113 // In case of improper error implementation or fatal exceptions
114 cause.addSuppressed(e)
115 handleCoroutineException(context, cause)
116 }
117 }
118 }
119
disposenull120 override fun dispose() {
121 disposed = true
122 cancel()
123 }
124
isDisposednull125 override fun isDisposed(): Boolean = disposed
126 }
127
128 /**
129 * @suppress
130 */
131 @Deprecated(
132 message = "CoroutineScope.mono is deprecated in favour of top-level mono",
133 level = DeprecationLevel.HIDDEN,
134 replaceWith = ReplaceWith("mono(context, block)")
135 ) // Since 1.3.0, will be error in 1.3.1 and hidden in 1.4.0
136 public fun <T> CoroutineScope.mono(
137 context: CoroutineContext = EmptyCoroutineContext,
138 block: suspend CoroutineScope.() -> T?
139 ): Mono<T> = monoInternal(this, context, block)
140
141 /**
142 * This is a lint function that was added already deprecated in order to guard against confusing usages on [Mono].
143 * On [Publisher] instances other than [Mono], this function is not deprecated.
144 *
145 * Both [awaitFirst] and [awaitSingle] await the first value, or throw [NoSuchElementException] if there is none, but
146 * the name [Mono.awaitSingle] better reflects the semantics of [Mono].
147 *
148 * For example, consider this code:
149 * ```
150 * myDbClient.findById(uniqueId).awaitFirst() // findById returns a `Mono`
151 * ```
152 * It looks like more than one value could be returned from `findById` and [awaitFirst] discards the extra elements,
153 * when in fact, at most a single value can be present.
154 *
155 * @suppress
156 */
157 @Deprecated(
158 message = "Mono produces at most one value, so the semantics of dropping the remaining elements are not useful. " +
159 "Please use awaitSingle() instead.",
160 level = DeprecationLevel.ERROR,
161 replaceWith = ReplaceWith("this.awaitSingle()")
162 ) // Warning since 1.5, error in 1.6
163 public suspend fun <T> Mono<T>.awaitFirst(): T = awaitSingle()
164
165 /**
166 * This is a lint function that was added already deprecated in order to guard against confusing usages on [Mono].
167 * On [Publisher] instances other than [Mono], this function is not deprecated.
168 *
169 * Both [awaitFirstOrDefault] and [awaitSingleOrNull] await the first value, or return some special value if there
170 * is none, but the name [Mono.awaitSingleOrNull] better reflects the semantics of [Mono].
171 *
172 * For example, consider this code:
173 * ```
174 * myDbClient.findById(uniqueId).awaitFirstOrDefault(default) // findById returns a `Mono`
175 * ```
176 * It looks like more than one value could be returned from `findById` and [awaitFirstOrDefault] discards the extra
177 * elements, when in fact, at most a single value can be present.
178 *
179 * @suppress
180 */
181 @Deprecated(
182 message = "Mono produces at most one value, so the semantics of dropping the remaining elements are not useful. " +
183 "Please use awaitSingleOrNull() instead.",
184 level = DeprecationLevel.ERROR,
185 replaceWith = ReplaceWith("this.awaitSingleOrNull() ?: default")
186 ) // Warning since 1.5, error in 1.6
187 public suspend fun <T> Mono<T>.awaitFirstOrDefault(default: T): T = awaitSingleOrNull() ?: default
188
189 /**
190 * This is a lint function that was added already deprecated in order to guard against confusing usages on [Mono].
191 * On [Publisher] instances other than [Mono], this function is not deprecated.
192 *
193 * Both [awaitFirstOrNull] and [awaitSingleOrNull] await the first value, or return some special value if there
194 * is none, but the name [Mono.awaitSingleOrNull] better reflects the semantics of [Mono].
195 *
196 * For example, consider this code:
197 * ```
198 * myDbClient.findById(uniqueId).awaitFirstOrNull() // findById returns a `Mono`
199 * ```
200 * It looks like more than one value could be returned from `findById` and [awaitFirstOrNull] discards the extra
201 * elements, when in fact, at most a single value can be present.
202 *
203 * @suppress
204 */
205 @Deprecated(
206 message = "Mono produces at most one value, so the semantics of dropping the remaining elements are not useful. " +
207 "Please use awaitSingleOrNull() instead.",
208 level = DeprecationLevel.ERROR,
209 replaceWith = ReplaceWith("this.awaitSingleOrNull()")
210 ) // Warning since 1.5, error in 1.6
211 public suspend fun <T> Mono<T>.awaitFirstOrNull(): T? = awaitSingleOrNull()
212
213 /**
214 * This is a lint function that was added already deprecated in order to guard against confusing usages on [Mono].
215 * On [Publisher] instances other than [Mono], this function is not deprecated.
216 *
217 * Both [awaitFirstOrElse] and [awaitSingleOrNull] await the first value, or return some special value if there
218 * is none, but the name [Mono.awaitSingleOrNull] better reflects the semantics of [Mono].
219 *
220 * For example, consider this code:
221 * ```
222 * myDbClient.findById(uniqueId).awaitFirstOrElse(defaultValue) // findById returns a `Mono`
223 * ```
224 * It looks like more than one value could be returned from `findById` and [awaitFirstOrElse] discards the extra
225 * elements, when in fact, at most a single value can be present.
226 *
227 * @suppress
228 */
229 @Deprecated(
230 message = "Mono produces at most one value, so the semantics of dropping the remaining elements are not useful. " +
231 "Please use awaitSingleOrNull() instead.",
232 level = DeprecationLevel.ERROR,
233 replaceWith = ReplaceWith("this.awaitSingleOrNull() ?: defaultValue()")
234 ) // Warning since 1.5, error in 1.6
235 public suspend fun <T> Mono<T>.awaitFirstOrElse(defaultValue: () -> T): T = awaitSingleOrNull() ?: defaultValue()
236
237 /**
238 * This is a lint function that was added already deprecated in order to guard against confusing usages on [Mono].
239 * On [Publisher] instances other than [Mono], this function is not deprecated.
240 *
241 * Both [awaitLast] and [awaitSingle] await the single value, or throw [NoSuchElementException] if there is none, but
242 * the name [Mono.awaitSingle] better reflects the semantics of [Mono].
243 *
244 * For example, consider this code:
245 * ```
246 * myDbClient.findById(uniqueId).awaitLast() // findById returns a `Mono`
247 * ```
248 * It looks like more than one value could be returned from `findById` and [awaitLast] discards the initial elements,
249 * when in fact, at most a single value can be present.
250 *
251 * @suppress
252 */
253 @Deprecated(
254 message = "Mono produces at most one value, so the last element is the same as the first. " +
255 "Please use awaitSingle() instead.",
256 level = DeprecationLevel.ERROR,
257 replaceWith = ReplaceWith("this.awaitSingle()")
258 ) // Warning since 1.5, error in 1.6
259 public suspend fun <T> Mono<T>.awaitLast(): T = awaitSingle()
260