Add the missing type boundaries to the reactive integrations

reactive/kotlinx-coroutines-jdk9/src/Publish.kt

@@ -29,7 +29,7 @@ import org.reactivestreams.FlowAdapters
  *        to cancellation and error handling may change in the future.
  */
 @ExperimentalCoroutinesApi // Since 1.3.x
-public fun <T> flowPublish(
+public fun <T : Any> flowPublish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Flow.Publisher<T> {

reactive/kotlinx-coroutines-reactive/src/Convert.kt

@@ -4,6 +4,7 @@
 
 package kotlinx.coroutines.reactive
 
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import org.reactivestreams.*
 import kotlin.coroutines.*
@@ -18,7 +19,10 @@ import kotlin.coroutines.*
 @Deprecated(message = "Deprecated in the favour of consumeAsFlow()",
     level = DeprecationLevel.WARNING, // Error in 1.4
     replaceWith = ReplaceWith("this.consumeAsFlow().asPublisher()"))
-public fun <T> ReceiveChannel<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): Publisher<T> = publish(context) {
-    for (t in this@asPublisher)
-        send(t)
-}
+public fun <T> ReceiveChannel<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): Publisher<T> =
+    // we call the deprecated version here because the non-deprecated one requires <T: Any> now
+    @Suppress("DEPRECATION_ERROR")
+    GlobalScope.publish(context) {
+        for (t in this@asPublisher)
+            send(t)
+    }

reactive/kotlinx-coroutines-reactive/src/Publish.kt

@@ -34,7 +34,7 @@ import kotlin.internal.*
  *        to cancellation and error handling may change in the future.
  */
 @ExperimentalCoroutinesApi
-public fun <T> publish(
+public fun <T : Any> publish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Publisher<T> {

reactive/kotlinx-coroutines-reactor/src/Convert.kt

@@ -36,7 +36,7 @@ public fun Job.asMono(context: CoroutineContext): Mono<Unit> = mono(context) { t
  * @param context -- the coroutine context from which the resulting mono is going to be signalled
  */
 @ExperimentalCoroutinesApi
-public fun <T> Deferred<T?>.asMono(context: CoroutineContext): Mono<T> = mono(context) { [email protected]() }
+public fun <T : Any> Deferred<T?>.asMono(context: CoroutineContext): Mono<T> = mono(context) { [email protected]() }
 
 /**
  * Converts a stream of elements received from the channel to the hot reactive flux.
@@ -48,7 +48,10 @@ public fun <T> Deferred<T?>.asMono(context: CoroutineContext): Mono<T> = mono(co
 @Deprecated(message = "Deprecated in the favour of consumeAsFlow()",
     level = DeprecationLevel.WARNING,
     replaceWith = ReplaceWith("this.consumeAsFlow().asFlux()"))
-public fun <T> ReceiveChannel<T>.asFlux(context: CoroutineContext = EmptyCoroutineContext): Flux<T> = flux(context) {
-    for (t in this@asFlux)
-        send(t)
-}
+public fun <T> ReceiveChannel<T>.asFlux(context: CoroutineContext = EmptyCoroutineContext): Flux<T> =
+    // use the deprecated version of `flux` here because the proper one has the <T : Any> boundary now
+    @Suppress("DEPRECATION_ERROR")
+    GlobalScope.flux(context) {
+        for (t in this@asFlux)
+            send(t)
+    }

reactive/kotlinx-coroutines-reactor/src/Flux.kt

@@ -34,7 +34,7 @@ import kotlin.internal.*
  *        to cancellation and error handling may change in the future.
  */
 @ExperimentalCoroutinesApi
-public fun <T> flux(
+public fun <T : Any> flux(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Flux<T> {

reactive/kotlinx-coroutines-reactor/src/Mono.kt

@@ -23,7 +23,7 @@ import kotlin.internal.*
  *
  * Method throws [IllegalArgumentException] if provided [context] contains a [Job] instance.
  */
-public fun <T> mono(
+public fun <T : Any> mono(
     context: CoroutineContext = EmptyCoroutineContext,
     block: suspend CoroutineScope.() -> T?
 ): Mono<T> {

reactive/kotlinx-coroutines-reactor/src/ReactorFlow.kt

@@ -25,7 +25,7 @@ import kotlin.coroutines.*
  * is used, so calls are performed from an arbitrary thread.
  */
 @JvmOverloads // binary compatibility
-public fun <T: Any> Flow<T>.asFlux(context: CoroutineContext = EmptyCoroutineContext): Flux<T> =
+public fun <T : Any> Flow<T>.asFlux(context: CoroutineContext = EmptyCoroutineContext): Flux<T> =
     FlowAsFlux(this, Dispatchers.Unconfined + context)
 
 private class FlowAsFlux<T : Any>(

reactive/kotlinx-coroutines-reactor/test/FluxSingleTest.kt

@@ -92,7 +92,7 @@ class FluxSingleTest : TestBase() {
 
     @Test
     fun testAwaitSingleOrNull() {
-        val flux = flux<String?> {
+        val flux = flux {
             send(Flux.empty<String>().awaitSingleOrNull() ?: "OK")
         }
 
@@ -169,7 +169,7 @@ class FluxSingleTest : TestBase() {
 
     @Test
     fun testAwaitFirstOrNull() {
-        val flux = flux<String?> {
+        val flux = flux {
             send(Flux.empty<String>().awaitFirstOrNull() ?: "OK")
         }
 

reactive/kotlinx-coroutines-reactor/test/ReactorContextTest.kt

@@ -74,7 +74,7 @@ class ReactorContextTest : TestBase() {
     }
 
 
-    private fun createFlux(): Flux<String?> = flux {
+    private fun createFlux(): Flux<String> = flux {
         val ctx = reactorContext()
         (1..3).forEach { send(ctx.getOrDefault(it, "noValue")) }
     }

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

@@ -44,7 +44,7 @@ public fun Job.asCompletable(context: CoroutineContext): Completable = rxComplet
  * @param context -- the coroutine context from which the resulting maybe is going to be signalled
  */
 @ExperimentalCoroutinesApi
-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) {
     [email protected]()
 }
 

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

@@ -20,7 +20,7 @@ import kotlin.internal.*
  * 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?
 ): Maybe<T> {

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

@@ -44,7 +44,7 @@ public fun Job.asCompletable(context: CoroutineContext): Completable = rxComplet
  * @param context -- the coroutine context from which the resulting maybe is going to be signalled
  */
 @ExperimentalCoroutinesApi
-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) {
     [email protected]()
 }
 

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

@@ -17,7 +17,7 @@ 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?
 ): Maybe<T> {