Update code to prepare for nullness annotations in rxjava3 and Guava.

(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.)

Guava's current nullness annotations likewise don't yet trigger Kotlin
compile errors. However, Guava did recently remove the misleading
`@Nullable` annotation from the parameter of `FutureCallback.onSuccess`:

Before:
https://github.com/google/guava/blob/v28.0/guava/src/com/google/common/util/concurrent/FutureCallback.java#L34

After:
https://github.com/google/guava/blob/v31.0.1/guava/src/com/google/common/util/concurrent/FutureCallback.java#L35

That means that a `FutureCallback<T>` can now implement `onSuccess(T)`
instead of `onSuccess(T?)`. And with a future change to Guava, it will
_have to_.

When I updated this project from Guava 28.0 to Guava 31.0.1, I found
that that pulled in a newer version of the Checker Framework nullness
annotations. That version was build with a newer version of Gradle, so
it generates Gradle module metadata:
https://docs.gradle.org/current/userguide/publishing_gradle_module_metadata.html

That metadata declares that the Checker Framework annotations require
Java 1.8. Even the old version had in fact required Java 1.8 -- and so
had even the old version of Guava -- just without having that encoded in
its metadata. So I fixed this by setting targetCompatibility and
sourceCompatibility to 1.8 for kotlinx-coroutines-guava, too.

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

@@ -168,6 +168,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"
     }
 }
 

integration/kotlinx-coroutines-guava/build.gradle.kts

@@ -2,12 +2,17 @@
  * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-val guavaVersion = "28.0-jre"
+val guavaVersion = "31.0.1-jre"
 
 dependencies {
     compile("com.google.guava:guava:$guavaVersion")
 }
 
+java {
+    targetCompatibility = JavaVersion.VERSION_1_8
+    sourceCompatibility = JavaVersion.VERSION_1_8
+}
+
 externalDocumentationLink(
     url = "https://google.github.io/guava/releases/$guavaVersion/api/docs/"
 )

integration/kotlinx-coroutines-guava/src/ListenableFuture.kt

@@ -133,10 +133,8 @@ public fun <T> ListenableFuture<T>.asDeferred(): Deferred<T> {
     // Finally, if this isn't done yet, attach a Listener that will complete the Deferred.
     val deferred = CompletableDeferred<T>()
     Futures.addCallback(this, object : FutureCallback<T> {
-        override fun onSuccess(result: T?) {
-            // Here we work with flexible types, so we unchecked cast to trick the type system
-            @Suppress("UNCHECKED_CAST")
-            runCatching { deferred.complete(result as T) }
+        override fun onSuccess(result: T) {
+            runCatching { deferred.complete(result) }
                 .onFailure { handleCoroutineException(EmptyCoroutineContext, it) }
         }
 
@@ -348,7 +346,7 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
      *
      * To preserve Coroutine's [CancellationException], this future points to either `T` or [Cancelled].
      */
-    private val auxFuture = SettableFuture.create<Any>()
+    private val auxFuture = SettableFuture.create<Any?>()
 
     /**
      * `true` if [auxFuture.get][ListenableFuture.get] throws [ExecutionException].
@@ -433,7 +431,7 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
     }
 
     /** See [get()]. */
-    private fun getInternal(result: Any): T = if (result is Cancelled) {
+    private fun getInternal(result: Any?): T = if (result is Cancelled) {
         throw CancellationException().initCause(result.exception)
     } else {
         // We know that `auxFuture` can contain either `T` or `Cancelled`.

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.WARNING,
     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.WARNING,
     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)
         }