Leverage definitely non-nullable types

Author: qwwdfsad

build.gradle

@@ -153,7 +153,6 @@ configure(subprojects.findAll { !sourceless.contains(it.name) }) {
         kotlinOptions.freeCompilerArgs += optInAnnotations.collect { "-Xopt-in=" + it }
         kotlinOptions.freeCompilerArgs += "-progressive"
         // Disable KT-36770 for RxJava2 integration
-        kotlinOptions.freeCompilerArgs += "-XXLanguage:-ProhibitUsingNullableTypeParameterAgainstNotNullAnnotated"
         // Remove null assertions to get smaller bytecode on Android
         kotlinOptions.freeCompilerArgs += ["-Xno-param-assertions", "-Xno-receiver-assertions", "-Xno-call-assertions"]
     }

gradle.properties

@@ -5,7 +5,7 @@
 # Kotlin
 version=1.5.2-SNAPSHOT
 group=org.jetbrains.kotlinx
-kotlin_version=1.5.30
+kotlin_version=1.6.0-RC-512
 
 # Dependencies
 junit_version=4.12

reactive/kotlinx-coroutines-rx2/build.gradle

@@ -18,6 +18,19 @@ 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

@@ -5,11 +5,8 @@
 package kotlinx.coroutines.rx2
 
 import io.reactivex.*
-import io.reactivex.disposables.Disposable
-import kotlinx.coroutines.CancellableContinuation
-import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.suspendCancellableCoroutine
+import io.reactivex.disposables.*
+import kotlinx.coroutines.*
 import kotlin.coroutines.*
 
 // ------------------------ CompletableSource ------------------------
@@ -43,12 +40,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>)
 }
 
 /**
@@ -118,11 +115,11 @@ 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> {
+    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 ------------------------
@@ -215,7 +212,7 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
     mode: Mode,
     default: T? = null
 ): T = suspendCancellableCoroutine { cont ->
-    subscribe(object : Observer<T> {
+    subscribe(object : Observer<T & Any> {
         private lateinit var subscription: Disposable
         private var value: T? = null
         private var seenValue = false
@@ -225,7 +222,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) {
@@ -266,6 +263,6 @@ private suspend fun <T> ObservableSource<T>.awaitOne(
         override fun onError(e: Throwable) {
             cont.resumeWithException(e)
         }
-    })
+    } as Observer<T>)
 }
 

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

@@ -23,7 +23,7 @@ import kotlinx.coroutines.reactive.*
 @Deprecated(message = "Deprecated in the favour of Flow", level = DeprecationLevel.ERROR) // Will be hidden in 1.5
 public fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    subscribe(channel as MaybeObserver<T>)
     return channel
 }
 
@@ -38,7 +38,7 @@ public fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
 @Deprecated(message = "Deprecated in the favour of Flow", level = DeprecationLevel.ERROR) // Will be hidden in 1.5
 public fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    subscribe(channel as Observer<T>)
     return channel
 }
 
@@ -63,20 +63,20 @@ public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit):
 @PublishedApi
 internal fun <T> MaybeSource<T>.toChannel(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    subscribe(channel as MaybeObserver<T>)
     return channel
 }
 
 @PublishedApi
 internal fun <T> ObservableSource<T>.toChannel(): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>()
-    subscribe(channel)
+    subscribe(channel as Observer<T>)
     return channel
 }
 
 @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)
 
@@ -90,12 +90,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
     }