Update code to prepare for nullness annotations in rxjava3.

(In addition to keeping the code compiling in the future, this change
should make some runtime nullness errors impossible.)

rxjava3 nullness annotations don't yet trigger Kotlin compile errors,
but that will be changing in Kotlin 1.7:
https://github.com/JetBrains/kotlin/blob/05822c59b516b6d252bd6d27e9032e660e15b625/core/compiler.common.jvm/src/org/jetbrains/kotlin/load/java/JavaNullabilityAnnotationSettings.kt#L42-L46

We can preview the behavior by passing
[email protected]:strict:
https://kotlinlang.org/docs/java-interop.html#nullability-annotations

We additionally set -Xtype-enhancement-improvements-strict-mode so that
the Kotlin compiler looks at type-use annotations on type arguments,
type parameters, etc.:
https://kotlinlang.org/docs/java-interop.html#annotating-type-arguments-and-type-parameters

Usually, the required update is to restrict a type parameter to
non-nullable types, since most rxjava types do not support null type
arguments. In a few cases, the update is to change an unnecessarily
nullable type to be non-nullable. Finally, I removed a `value == null`
check in `RxMaybeCoroutine.onCompleted` that the compiler now identifies
as unnecessary. (I can keep it if you'd prefer.)

I should warn you that I have very little understanding of coroutines
and of this library. I'm here because we're seeing compile errors inside
Google as we work to improve how we handle Kotlin-Java interoperability,
and these changes looked like they might be the right fixes. Sorry for
any mistakes.

Author: cpovirk

build.gradle

@@ -174,6 +174,10 @@ configure(subprojects.findAll { !sourceless.contains(it.name) }) {
         kotlinOptions.freeCompilerArgs += "-XXLanguage:-ProhibitUsingNullableTypeParameterAgainstNotNullAnnotated"
         // Remove null assertions to get smaller bytecode on Android
         kotlinOptions.freeCompilerArgs += ["-Xno-param-assertions", "-Xno-receiver-assertions", "-Xno-call-assertions"]
+        // Recognize rxjava3 nullness annotations even before that becomes the default (which will happen in 1.7): https://kotlinlang.org/docs/java-interop.html#nullability-annotations
+        kotlinOptions.freeCompilerArgs += "[email protected]:strict"
+        // Recognize nullness annotations on type arguments, etc.: https://kotlinlang.org/docs/java-interop.html#annotating-type-arguments-and-type-parameters
+        kotlinOptions.freeCompilerArgs += "-Xtype-enhancement-improvements-strict-mode"
     }
 }
 

reactive/kotlinx-coroutines-rx3/src/RxAwait.kt

@@ -42,7 +42,7 @@ public suspend fun CompletableSource.await(): Unit = suspendCancellableCoroutine
  * function immediately resumes with [CancellationException] and disposes of its subscription.
  */
 @Suppress("UNCHECKED_CAST")
-public suspend fun <T> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
+public suspend fun <T : Any> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
     subscribe(object : MaybeObserver<T> {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCancellation(d) }
         override fun onComplete() { cont.resume(null) }
@@ -61,7 +61,7 @@ public suspend fun <T> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellab
  *
  * @throws NoSuchElementException if no elements were produced by this [MaybeSource].
  */
-public suspend fun <T> MaybeSource<T>.awaitSingle(): T = awaitSingleOrNull() ?: throw NoSuchElementException()
+public suspend fun <T : Any> MaybeSource<T>.awaitSingle(): T = awaitSingleOrNull() ?: throw NoSuchElementException()
 
 /**
  * Awaits for completion of the maybe without blocking a thread.
@@ -84,7 +84,7 @@ public suspend fun <T> MaybeSource<T>.awaitSingle(): T = awaitSingleOrNull() ?:
     level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("this.awaitSingleOrNull()")
 ) // Warning since 1.5, error in 1.6, hidden in 1.7
-public suspend fun <T> MaybeSource<T>.await(): T? = awaitSingleOrNull()
+public suspend fun <T : Any> MaybeSource<T>.await(): T? = awaitSingleOrNull()
 
 /**
  * Awaits for completion of the maybe without blocking a thread.
@@ -107,7 +107,7 @@ public suspend fun <T> MaybeSource<T>.await(): T? = awaitSingleOrNull()
     level = DeprecationLevel.ERROR,
     replaceWith = ReplaceWith("this.awaitSingleOrNull() ?: default")
 ) // Warning since 1.5, error in 1.6, hidden in 1.7
-public suspend fun <T> MaybeSource<T>.awaitOrDefault(default: T): T = awaitSingleOrNull() ?: default
+public suspend fun <T : Any> MaybeSource<T>.awaitOrDefault(default: T): T = awaitSingleOrNull() ?: default
 
 // ------------------------ SingleSource ------------------------
 
@@ -119,7 +119,7 @@ public suspend fun <T> MaybeSource<T>.awaitOrDefault(default: T): T = awaitSingl
  * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
-public suspend fun <T> SingleSource<T>.await(): T = suspendCancellableCoroutine { cont ->
+public suspend fun <T : Any> SingleSource<T>.await(): T = suspendCancellableCoroutine { cont ->
     subscribe(object : SingleObserver<T> {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCancellation(d) }
         override fun onSuccess(t: T) { cont.resume(t) }
@@ -139,7 +139,7 @@ public suspend fun <T> SingleSource<T>.await(): T = suspendCancellableCoroutine
  *
  * @throws NoSuchElementException if the observable does not emit any value
  */
-public suspend fun <T> ObservableSource<T>.awaitFirst(): T = awaitOne(Mode.FIRST)
+public suspend fun <T : Any> ObservableSource<T>.awaitFirst(): T = awaitOne(Mode.FIRST)
 
 /**
  * Awaits the first value from the given [Observable], or returns the [default] value if none is emitted, without
@@ -150,7 +150,7 @@ public suspend fun <T> ObservableSource<T>.awaitFirst(): T = awaitOne(Mode.FIRST
  * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
-public suspend fun <T> ObservableSource<T>.awaitFirstOrDefault(default: T): T = awaitOne(Mode.FIRST_OR_DEFAULT, default)
+public suspend fun <T : Any> ObservableSource<T>.awaitFirstOrDefault(default: T): T = awaitOne(Mode.FIRST_OR_DEFAULT, default)
 
 /**
  * Awaits the first value from the given [Observable], or returns `null` if none is emitted, without blocking the
@@ -161,7 +161,7 @@ public suspend fun <T> ObservableSource<T>.awaitFirstOrDefault(default: T): T =
  * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
-public suspend fun <T> ObservableSource<T>.awaitFirstOrNull(): T? = awaitOne(Mode.FIRST_OR_DEFAULT)
+public suspend fun <T : Any> ObservableSource<T>.awaitFirstOrNull(): T? = awaitOne(Mode.FIRST_OR_DEFAULT)
 
 /**
  * Awaits the first value from the given [Observable], or calls [defaultValue] to get a value if none is emitted,
@@ -172,7 +172,7 @@ public suspend fun <T> ObservableSource<T>.awaitFirstOrNull(): T? = awaitOne(Mod
  * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
-public suspend fun <T> ObservableSource<T>.awaitFirstOrElse(defaultValue: () -> T): T =
+public suspend fun <T : Any> ObservableSource<T>.awaitFirstOrElse(defaultValue: () -> T): T =
     awaitOne(Mode.FIRST_OR_DEFAULT) ?: defaultValue()
 
 /**
@@ -185,7 +185,7 @@ public suspend fun <T> ObservableSource<T>.awaitFirstOrElse(defaultValue: () ->
  *
  * @throws NoSuchElementException if the observable does not emit any value
  */
-public suspend fun <T> ObservableSource<T>.awaitLast(): T = awaitOne(Mode.LAST)
+public suspend fun <T : Any> ObservableSource<T>.awaitLast(): T = awaitOne(Mode.LAST)
 
 /**
  * Awaits the single value from the given observable without blocking the thread and returns the resulting value, or,
@@ -198,7 +198,7 @@ public suspend fun <T> ObservableSource<T>.awaitLast(): T = awaitOne(Mode.LAST)
  * @throws NoSuchElementException if the observable does not emit any value
  * @throws IllegalArgumentException if the observable emits more than one value
  */
-public suspend fun <T> ObservableSource<T>.awaitSingle(): T = awaitOne(Mode.SINGLE)
+public suspend fun <T : Any> ObservableSource<T>.awaitSingle(): T = awaitOne(Mode.SINGLE)
 
 // ------------------------ private ------------------------
 
@@ -213,7 +213,7 @@ private enum class Mode(val s: String) {
     override fun toString(): String = s
 }
 
-private suspend fun <T> ObservableSource<T>.awaitOne(
+private suspend fun <T : Any> ObservableSource<T>.awaitOne(
     mode: Mode,
     default: T? = null
 ): T = suspendCancellableCoroutine { cont ->

reactive/kotlinx-coroutines-rx3/src/RxChannel.kt

@@ -19,7 +19,7 @@ import kotlinx.coroutines.flow.*
  * [MaybeSource] doesn't have a corresponding [Flow] adapter, so it should be transformed to [Observable] first.
  */
 @PublishedApi
-internal fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
+internal fun <T : Any> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
     subscribe(channel)
     return channel
@@ -33,7 +33,7 @@ internal fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
  * [ObservableSource] doesn't have a corresponding [Flow] adapter, so it should be transformed to [Observable] first.
  */
 @PublishedApi
-internal fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
+internal fun <T : Any> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
     subscribe(channel)
     return channel
@@ -45,7 +45,7 @@ internal fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
  * If [action] throws an exception at some point or if the [MaybeSource] raises an error, the exception is rethrown from
  * [collect].
  */
-public suspend inline fun <T> MaybeSource<T>.collect(action: (T) -> Unit): Unit =
+public suspend inline fun <T : Any> MaybeSource<T>.collect(action: (T) -> Unit): Unit =
     openSubscription().consumeEach(action)
 
 /**
@@ -54,11 +54,11 @@ public suspend inline fun <T> MaybeSource<T>.collect(action: (T) -> Unit): Unit
  * If [action] throws an exception at some point, the subscription is cancelled, and the exception is rethrown from
  * [collect]. Also, if the [ObservableSource] signals an error, that error is rethrown from [collect].
  */
-public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit): Unit =
+public suspend inline fun <T : Any> ObservableSource<T>.collect(action: (T) -> Unit): Unit =
     openSubscription().consumeEach(action)
 
 @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")
-private class SubscriptionChannel<T> :
+private class SubscriptionChannel<T : Any> :
     LinkedListChannel<T>(null), Observer<T>, MaybeObserver<T>
 {
     private val _subscription = atomic<Disposable?>(null)

reactive/kotlinx-coroutines-rx3/src/RxConvert.kt

@@ -42,7 +42,7 @@ public fun Job.asCompletable(context: CoroutineContext): Completable = rxComplet
  *
  * @param context -- the coroutine context from which the resulting maybe is going to be signalled
  */
-public fun <T> Deferred<T?>.asMaybe(context: CoroutineContext): Maybe<T> = rxMaybe(context) {
+public fun <T : Any> Deferred<T>.asMaybe(context: CoroutineContext): Maybe<T> = rxMaybe(context) {
     this@asMaybe.await()
 }
 

reactive/kotlinx-coroutines-rx3/src/RxMaybe.kt

@@ -17,33 +17,33 @@ import kotlin.coroutines.*
  * 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.
  */
-public fun <T> rxMaybe(
+public fun <T : Any> rxMaybe(
     context: CoroutineContext = EmptyCoroutineContext,
-    block: suspend CoroutineScope.() -> T?
+    block: suspend CoroutineScope.() -> T
 ): Maybe<T> {
     require(context[Job] === null) { "Maybe context cannot contain job in it." +
             "Its lifecycle should be managed via Disposable handle. Had $context" }
     return rxMaybeInternal(GlobalScope, context, block)
 }
 
-private fun <T> rxMaybeInternal(
+private fun <T : Any> rxMaybeInternal(
     scope: CoroutineScope, // support for legacy rxMaybe in scope
     context: CoroutineContext,
-    block: suspend CoroutineScope.() -> T?
+    block: suspend CoroutineScope.() -> T
 ): Maybe<T> = Maybe.create { subscriber ->
     val newContext = scope.newCoroutineContext(context)
     val coroutine = RxMaybeCoroutine(newContext, subscriber)
     subscriber.setCancellable(RxCancellable(coroutine))
     coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
 }
 
-private class RxMaybeCoroutine<T>(
+private class RxMaybeCoroutine<T : Any>(
     parentContext: CoroutineContext,
     private val subscriber: MaybeEmitter<T>
 ) : AbstractCoroutine<T>(parentContext, false, true) {
     override fun onCompleted(value: T) {
         try {
-            if (value == null) subscriber.onComplete() else subscriber.onSuccess(value)
+            subscriber.onSuccess(value)
         } catch (e: Throwable) {
             handleUndeliverableException(e, context)
         }