Skip to content

Commit 09fb524

Browse files
committed
Fixes
1 parent 7fc77c4 commit 09fb524

File tree

4 files changed

+9
-4
lines changed

4 files changed

+9
-4
lines changed

reactive/kotlinx-coroutines-reactor/src/Mono.kt

+2-1
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
package kotlinx.coroutines.reactor
88

99
import kotlinx.coroutines.*
10+
import kotlinx.coroutines.reactive.*
1011
import org.reactivestreams.*
1112
import reactor.core.*
1213
import reactor.core.publisher.*
@@ -43,7 +44,7 @@ public fun <T> mono(
4344
* function immediately cancels its [Subscription] and resumes with [CancellationException].
4445
*/
4546
public suspend fun <T> Mono<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
46-
subscribe(object : Subscriber<T> {
47+
injectCoroutineContext(cont.context).subscribe(object : Subscriber<T> {
4748
private var seenValue = false
4849

4950
override fun onSubscribe(s: Subscription) {

reactive/kotlinx-coroutines-reactor/test/FluxSingleTest.kt

+3-3
Original file line numberDiff line numberDiff line change
@@ -104,11 +104,11 @@ class FluxSingleTest : TestBase() {
104104
@Test
105105
fun testAwaitSingleOrNullException() {
106106
val flux = flux {
107-
send((Flux.just("!", "#").awaitSingleOrNull() ?: "O") + "K")
107+
send((Flux.just("O", "#").awaitSingleOrNull() ?: "!") + "K")
108108
}
109109

110-
checkSingleValue(flux) {
111-
assertEquals("OK", it)
110+
checkErroneous(flux) {
111+
assert(it is IllegalArgumentException)
112112
}
113113
}
114114

reactive/kotlinx-coroutines-rx2/api/kotlinx-coroutines-rx2.api

+2
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,9 @@ public final class kotlinx/coroutines/rx2/RxAwaitKt {
88
public static final fun awaitFirstOrNull (Lio/reactivex/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
99
public static final fun awaitLast (Lio/reactivex/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1010
public static final fun awaitOrDefault (Lio/reactivex/MaybeSource;Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
11+
public static final fun awaitSingle (Lio/reactivex/MaybeSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1112
public static final fun awaitSingle (Lio/reactivex/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
13+
public static final fun awaitSingleOrNull (Lio/reactivex/MaybeSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1214
}
1315

1416
public final class kotlinx/coroutines/rx2/RxChannelKt {

reactive/kotlinx-coroutines-rx3/api/kotlinx-coroutines-rx3.api

+2
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,9 @@ public final class kotlinx/coroutines/rx3/RxAwaitKt {
88
public static final fun awaitFirstOrNull (Lio/reactivex/rxjava3/core/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
99
public static final fun awaitLast (Lio/reactivex/rxjava3/core/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1010
public static final fun awaitOrDefault (Lio/reactivex/rxjava3/core/MaybeSource;Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
11+
public static final fun awaitSingle (Lio/reactivex/rxjava3/core/MaybeSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1112
public static final fun awaitSingle (Lio/reactivex/rxjava3/core/ObservableSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
13+
public static final fun awaitSingleOrNull (Lio/reactivex/rxjava3/core/MaybeSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1214
}
1315

1416
public final class kotlinx/coroutines/rx3/RxChannelKt {

0 commit comments

Comments
 (0)