Repository: Kotlin/kotlinx.coroutines
File: reactive/kotlinx-coroutines-rx2/src/RxAwait.kt (Kotlin)

Code before:
        }

        override fun onComplete() {
            cont.resume(Unit)
        }

        override fun onError(e: Throwable) {
            cont.resumeWithException(e)
        }
    })
}

// ------------------------ MaybeSource ------------------------

/**
 * Awaits for completion of the [MaybeSource] without blocking the thread.
 * Returns the resulting value, or `null` if no value is produced, or throws the corresponding exception if this
 * [MaybeSource] produces an error.
 *
 * This suspending function is cancellable.
 * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this
 * function immediately resumes with [CancellationException] and disposes of its subscription.
 */
@Suppress("UNCHECKED_CAST")
public suspend fun <T> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
    subscribe(object : MaybeObserver<T & Any> {
        override fun onSubscribe(d: Disposable) {
            cont.disposeOnCancellation(d)
        }

        override fun onComplete() {
            cont.resume(null)
        }

        override fun onSuccess(t: T & Any) {
            cont.resume(t)
        }

        override fun onError(error: Throwable) {
            cont.resumeWithException(error)
        }
    } as MaybeObserver<T>)
}

/**
 * Awaits for completion of the [MaybeSource] without blocking the thread.
 * Returns the resulting value, or throws if either no value is produced or this [MaybeSource] produces an error.
 *
 * This suspending function is cancellable.
 * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this
 * function immediately resumes with [CancellationException] and disposes of its subscription.

Review comment (suggestion):
```suggestion
    subscribe(object : MaybeObserver<T> {
```
Then, the cast can be removed, along with the suppression.

Code after:
        }

        override fun onComplete() {
            cont.resume(Unit)
        }

        override fun onError(e: Throwable) {
            cont.resumeWithException(e)
        }
    })
}

// ------------------------ MaybeSource ------------------------

/**
 * Awaits for completion of the [MaybeSource] without blocking the thread.
 * Returns the resulting value, or `null` if no value is produced, or throws the corresponding exception if this
 * [MaybeSource] produces an error.
 *
 * This suspending function is cancellable.
 * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this
 * function immediately resumes with [CancellationException] and disposes of its subscription.
 */
public suspend fun <T> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
    subscribe(object : MaybeObserver<T> {
        override fun onSubscribe(d: Disposable) {
            cont.disposeOnCancellation(d)
        }

        override fun onComplete() {
            cont.resume(null)
        }

        override fun onSuccess(t: T & Any) {
            cont.resume(t)
        }

        override fun onError(error: Throwable) {
            cont.resumeWithException(error)
        }
    })
}

/**
 * Awaits for completion of the [MaybeSource] without blocking the thread.
 * Returns the resulting value, or throws if either no value is produced or this [MaybeSource] produces an error.
 *
 * This suspending function is cancellable.
 * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this
 * function immediately resumes with [CancellationException] and disposes of its subscription.
 *