Update to newer Kotlin, update RxJava3 module

Author: qwwdfsad

build.gradle

@@ -169,7 +169,8 @@ configure(subprojects.findAll { !sourceless.contains(it.name) }) {
     tasks.withType(org.jetbrains.kotlin.gradle.tasks.AbstractKotlinCompile).all {
         kotlinOptions.freeCompilerArgs += OptInPresetKt.optInAnnotations.collect { "-Xopt-in=" + it }
         kotlinOptions.freeCompilerArgs += "-progressive"
-        // Disable KT-36770 for RxJava2 integration
+        // Internal implementation details of our Rx integration
+        kotlinOptions.freeCompilerArgs += "-XXLanguage:+DefinitelyNonNullableTypes"
         // Remove null assertions to get smaller bytecode on Android
         kotlinOptions.freeCompilerArgs += ["-Xno-param-assertions", "-Xno-receiver-assertions", "-Xno-call-assertions"]
     }

reactive/kotlinx-coroutines-rx2/build.gradle

@@ -18,19 +18,6 @@ tasks.withType(DokkaTaskPartial.class) {
         }
     }
 }
-tasks {
-    compileKotlin {
-        kotlinOptions {
-            languageVersion = "1.7"
-        }
-    }
-
-    compileTestKotlin {
-        kotlinOptions {
-            languageVersion = "1.7"
-        }
-    }
-}
 
 task testNG(type: Test) {
     useTestNG()

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

@@ -115,6 +115,7 @@ public suspend fun <T> MaybeSource<T>.awaitOrDefault(default: T): T = awaitSingl
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
 public suspend fun <T> SingleSource<T>.await(): T = suspendCancellableCoroutine { cont ->
+    @Suppress("UNCHECKED_CAST")
     subscribe(object : SingleObserver<T & Any> {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCancellation(d) }
         override fun onSuccess(t: T & Any) { cont.resume(t) }
@@ -212,6 +213,7 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
     mode: Mode,
     default: T? = null
 ): T = suspendCancellableCoroutine { cont ->
+    @Suppress("UNCHECKED_CAST")
     subscribe(object : Observer<T & Any> {
         private lateinit var subscription: Disposable
         private var value: T? = null

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

@@ -43,12 +43,12 @@ public suspend fun CompletableSource.await(): Unit = suspendCancellableCoroutine
  */
 @Suppress("UNCHECKED_CAST")
 public suspend fun <T> MaybeSource<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
-    subscribe(object : MaybeObserver<T> {
+    subscribe(object : MaybeObserver<T & Any> {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCancellation(d) }
         override fun onComplete() { cont.resume(null) }
-        override fun onSuccess(t: T) { cont.resume(t) }
+        override fun onSuccess(t: T & Any) { cont.resume(t) }
         override fun onError(error: Throwable) { cont.resumeWithException(error) }
-    })
+    } as MaybeObserver<T>)
 }
 
 /**
@@ -120,11 +120,12 @@ public suspend fun <T> MaybeSource<T>.awaitOrDefault(default: T): T = awaitSingl
  * function immediately disposes of its subscription and resumes with [CancellationException].
  */
 public suspend fun <T> SingleSource<T>.await(): T = suspendCancellableCoroutine { cont ->
-    subscribe(object : SingleObserver<T> {
+    @Suppress("UNCHECKED_CAST")
+    subscribe(object : SingleObserver<T & Any> {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCancellation(d) }
-        override fun onSuccess(t: T) { cont.resume(t) }
+        override fun onSuccess(t: T & Any) { cont.resume(t) }
         override fun onError(error: Throwable) { cont.resumeWithException(error) }
-    })
+    } as SingleObserver<T>)
 }
 
 // ------------------------ ObservableSource ------------------------
@@ -217,7 +218,8 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
     mode: Mode,
     default: T? = null
 ): T = suspendCancellableCoroutine { cont ->
-    subscribe(object : Observer<T> {
+    @Suppress("UNCHECKED_CAST")
+    subscribe(object : Observer<T & Any> {
         private lateinit var subscription: Disposable
         private var value: T? = null
         private var seenValue = false
@@ -227,7 +229,7 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
             cont.invokeOnCancellation { sub.dispose() }
         }
 
-        override fun onNext(t: T) {
+        override fun onNext(t: T & Any) {
             when (mode) {
                 Mode.FIRST, Mode.FIRST_OR_DEFAULT -> {
                     if (!seenValue) {
@@ -249,7 +251,6 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
             }
         }
 
-        @Suppress("UNCHECKED_CAST")
         override fun onComplete() {
             if (seenValue) {
                 if (cont.isActive) cont.resume(value as T)
@@ -268,6 +269,6 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
         override fun onError(e: Throwable) {
             cont.resumeWithException(e)
         }
-    })
+    } as Observer<T>)
 }
 

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

@@ -21,7 +21,8 @@ import kotlinx.coroutines.flow.*
 @PublishedApi
 internal fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    @Suppress("UNCHECKED_CAST")
+    subscribe(channel as MaybeObserver<T>)
     return channel
 }
 
@@ -35,7 +36,8 @@ internal fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
 @PublishedApi
 internal fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    @Suppress("UNCHECKED_CAST")
+    subscribe(channel as Observer<T>)
     return channel
 }
 
@@ -59,7 +61,7 @@ public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit):
 
 @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")
 private class SubscriptionChannel<T> :
-    LinkedListChannel<T>(null), Observer<T>, MaybeObserver<T>
+    LinkedListChannel<T>(null), Observer<T & Any>, MaybeObserver<T & Any>
 {
     private val _subscription = atomic<Disposable?>(null)
 
@@ -73,12 +75,12 @@ private class SubscriptionChannel<T> :
         _subscription.value = sub
     }
 
-    override fun onSuccess(t: T) {
+    override fun onSuccess(t: T & Any) {
         trySend(t)
         close(cause = null)
     }
 
-    override fun onNext(t: T) {
+    override fun onNext(t: T & Any) {
         trySend(t) // Safe to ignore return value here, expectedly racing with cancellation
     }