Upgrade to Coroutines 1.3.0-RC

This commit upgrades Coroutines support to kotlinx.coroutines
1.3.0-RC, leverages the new Coroutines BOM and refine Coroutines
detection to avoid false positives.

Only Coroutines to Mono context interoperability is supported
for now.

CLoses gh-23326

build.gradle

@@ -30,7 +30,7 @@ ext {
 	}
 
 	aspectjVersion       = "1.9.4"
-	coroutinesVersion    = "1.3.0-M2"
+	coroutinesVersion    = "1.3.0-RC"
 	freemarkerVersion    = "2.3.28"
 	groovyVersion        = "2.5.7"
 	hsqldbVersion        = "2.5.0"
@@ -79,6 +79,7 @@ configure(allprojects) { project ->
 		imports {
 			mavenBom "org.junit:junit-bom:${junit5Version}"
 			mavenBom "org.jetbrains.kotlin:kotlin-bom:${kotlinVersion}"
+			mavenBom "org.jetbrains.kotlinx:kotlinx-coroutines-bom:${coroutinesVersion}"
 		}
 	}
 

spring-core-coroutines/spring-core-coroutines.gradle

@@ -4,8 +4,8 @@ dependencies {
 	compile("org.jetbrains.kotlin:kotlin-reflect")
 	compile("org.jetbrains.kotlin:kotlin-stdlib")
 	compile("io.projectreactor:reactor-core")
-	compile("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
-	compile("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
+	compile("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+	compile("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
 }
 
 eclipse {

spring-core-coroutines/src/main/kotlin/org/springframework/core/CoroutinesUtils.kt

@@ -40,7 +40,7 @@ import kotlin.reflect.jvm.kotlinFunction
  * @since 5.2
  */
 internal fun <T: Any> deferredToMono(source: Deferred<T>) =
-		GlobalScope.mono(Dispatchers.Unconfined) { source.await() }
+		mono(Dispatchers.Unconfined) { source.await() }
 
 /**
  * Convert a [Mono] instance to a [Deferred] one.
@@ -63,7 +63,7 @@ internal fun <T: Any> monoToDeferred(source: Mono<T>) =
 internal fun invokeHandlerMethod(method: Method, bean: Any, vararg args: Any?): Any? {
 	val function = method.kotlinFunction!!
 	return if (function.isSuspend) {
-		val mono = GlobalScope.mono(Dispatchers.Unconfined) {
+		val mono = mono(Dispatchers.Unconfined) {
 			function.callSuspend(bean, *args.sliceArray(0..(args.size-2)))
 					.let { if (it == Unit) null else it }
 		}.onErrorMap(InvocationTargetException::class.java) { it.targetException }

spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java

@@ -47,7 +47,7 @@ import org.springframework.util.ReflectionUtils;
  *
  * <p>By default, depending on classpath availability, adapters are registered
  * for Reactor, RxJava 1, RxJava 2 types, {@link CompletableFuture}, Java 9+
- * {@code Flow.Publisher} and Kotlin Coroutines {@code Deferred}.
+ * {@code Flow.Publisher} and Kotlin Coroutines {@code Deferred} and {@code Flow}.
  *
  * @author Rossen Stoyanchev
  * @author Sebastien Deleuze
@@ -97,13 +97,9 @@ public class ReactiveAdapterRegistry {
 		// We can fall back on "reactive-streams-flow-bridge" (once released)
 
 		// Coroutines
-		if (this.reactorPresent && ClassUtils.isPresent("kotlinx.coroutines.Deferred", classLoader)) {
+		if (this.reactorPresent && ClassUtils.isPresent("kotlinx.coroutines.reactor.MonoKt", classLoader) && ClassUtils.isPresent("kotlinx.coroutines.reactive.flow.PublisherAsFlowKt", classLoader)) {
 			new CoroutinesRegistrar().registerAdapters(this);
 		}
-		// TODO Use a single CoroutinesRegistrar when Flow will be not experimental anymore
-		if (this.reactorPresent && ClassUtils.isPresent("kotlinx.coroutines.flow.Flow", classLoader)) {
-			new CoroutinesFlowRegistrar().registerAdapters(this);
-		}
 	}
 
 
@@ -353,13 +349,7 @@ public class ReactiveAdapterRegistry {
 							() -> CompletableDeferredKt.CompletableDeferred(null)),
 					source -> CoroutinesUtils.deferredToMono((Deferred<?>) source),
 					source -> CoroutinesUtils.monoToDeferred(Mono.from(source)));
-		}
-	}
 
-
-	private static class CoroutinesFlowRegistrar {
-
-		void registerAdapters(ReactiveAdapterRegistry registry) {
 			registry.registerReactiveType(
 					ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, FlowKt::emptyFlow),
 					source -> FlowAsPublisherKt.from((kotlinx.coroutines.flow.Flow<?>) source),
@@ -367,5 +357,4 @@ public class ReactiveAdapterRegistry {
 			);
 		}
 	}
-
 }

spring-messaging/spring-messaging.gradle

@@ -18,8 +18,8 @@ dependencies {
 	optional("io.rsocket:rsocket-transport-netty:${rsocketVersion}")
 	optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}")
 	optional("javax.xml.bind:jaxb-api:2.3.1")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
 	testCompile("javax.inject:javax.inject-tck:1")
 	testCompile("javax.servlet:javax.servlet-api:4.0.1")
 	testCompile("javax.validation:validation-api:1.1.0.Final")

spring-test/spring-test.gradle

@@ -53,8 +53,8 @@ dependencies {
 	optional("org.jetbrains.kotlin:kotlin-reflect")
 	optional("org.jetbrains.kotlin:kotlin-stdlib")
 	optional("io.projectreactor:reactor-test")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
 	testCompile(project(":spring-context-support"))
 	testCompile(project(":spring-oxm"))
 	testCompile("javax.annotation:javax.annotation-api:1.3.2")

spring-webflux/spring-webflux.gradle

@@ -41,8 +41,8 @@ dependencies {
 	optional("org.jetbrains.kotlin:kotlin-reflect")
 	optional("org.jetbrains.kotlin:kotlin-stdlib")
 	optional("com.google.protobuf:protobuf-java-util:3.9.0")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
-	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+	optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
 	testCompile("javax.xml.bind:jaxb-api:2.3.1")
 	testCompile("com.fasterxml:aalto-xml:1.1.1")
 	testCompile("org.hibernate:hibernate-validator:6.0.17.Final")

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/CoRouterFunctionDsl.kt

@@ -17,7 +17,6 @@
 package org.springframework.web.reactive.function.server
 
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.reactor.mono
 import org.springframework.core.io.Resource
 import org.springframework.http.HttpMethod
@@ -421,7 +420,7 @@ class CoRouterFunctionDsl(private val init: (CoRouterFunctionDsl.() -> Unit)) {
 	 */
 	fun resources(lookupFunction: suspend (ServerRequest) -> Resource?) {
 		builder.resources {
-			GlobalScope.mono(Dispatchers.Unconfined) {
+			mono(Dispatchers.Unconfined) {
 				lookupFunction.invoke(it)
 			}
 		}
@@ -436,7 +435,7 @@ class CoRouterFunctionDsl(private val init: (CoRouterFunctionDsl.() -> Unit)) {
 	}
 
 	private fun asHandlerFunction(init: suspend (ServerRequest) -> ServerResponse) = HandlerFunction {
-		GlobalScope.mono(Dispatchers.Unconfined) {
+		mono(Dispatchers.Unconfined) {
 			init(it)
 		}
 	}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/server/ServerWebExchangeExtensions.kt

@@ -17,7 +17,6 @@
 package org.springframework.web.reactive.server
 
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.reactive.awaitSingle
 import kotlinx.coroutines.reactor.mono
 import org.springframework.http.codec.multipart.Part
@@ -69,4 +68,4 @@ suspend fun ServerWebExchange.awaitSession(): WebSession =
  * @since 5.2
  */
 fun ServerWebExchange.Builder.principal(supplier: suspend () -> Principal): ServerWebExchange.Builder
-        = principal(GlobalScope.mono(Dispatchers.Unconfined) { supplier.invoke() })
+        = principal(mono(Dispatchers.Unconfined) { supplier.invoke() })