Fixes

Author: dkhalanskyjb

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

@@ -7,22 +7,23 @@
 package kotlinx.coroutines.reactor
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.reactive.*
 import org.reactivestreams.*
 import reactor.core.*
 import reactor.core.publisher.*
 import kotlin.coroutines.*
 import kotlin.internal.*
 
 /**
- * Creates cold [mono][Mono] that runs a given [block] in a coroutine and emits its result.
+ * Creates a cold [mono][Mono] that runs a given [block] in a coroutine and emits its result.
  * Every time the returned mono is subscribed, it starts a new coroutine.
  * If the result of [block] is `null`, [MonoSink.success] is invoked without a value.
  * Unsubscribing cancels the running coroutine.
  *
  * Coroutine context can be specified with [context] argument.
  * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
  *
- * Method throws [IllegalArgumentException] if provided [context] contains a [Job] instance.
+ * @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
  */
 public fun <T> mono(
     context: CoroutineContext = EmptyCoroutineContext,
@@ -33,22 +34,6 @@ public fun <T> mono(
     return monoInternal(GlobalScope, context, block)
 }
 
-@Suppress("UNCHECKED_CAST")
-internal suspend fun <T> Mono<T>.await(): T? = (this as Mono<T?>).awaitOrDefault(null)
-
-internal suspend fun <T> Mono<T>.awaitOrDefault(default: T): T = suspendCancellableCoroutine { cont ->
-    subscribe(object: CoreSubscriber<T> {
-        override fun onSubscribe(s: Subscription) {
-            cont.invokeOnCancellation { s.cancel() }
-            s.request(1)
-        }
-        override fun onNext(t: T) { cont.resume(t) }
-        override fun onError(t: Throwable) { cont.resumeWithException(t) }
-        override fun onComplete() { cont.resume(default) }
-
-    })
-}
-
 @Deprecated(
     message = "CoroutineScope.mono is deprecated in favour of top-level mono",
     level = DeprecationLevel.ERROR,

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

@@ -333,7 +333,7 @@ class MonoTest : TestBase() {
         newSingleThreadContext("testDownstreamCancellationDoesNotThrow").use { pool ->
             val job = launch(pool, start = CoroutineStart.UNDISPATCHED) {
                 expect(1)
-                mono.await()
+                mono.awaitFirstOrNull()
             }
             launch(pool) {
                 job.cancelAndJoin()