root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

Use Dispatchers.Unconfined for Coroutines 

As of Coroutines 1.2.0-alpha, Dispatchers.Unconfined
is a stable API so we can leverage it in order to get
better performances in our Reactive to Coroutines
bridge.

See gh-19975
This commit is contained in:
Sebastien Deleuze 2019-03-31 10:56:05 +02:00
parent 3387d3ec89
commit beb491b840
4 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -18,6 +18,7 @@
package org.springframework.core
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import kotlinx.coroutines.reactive.awaitFirstOrNull
@ -36,7 +37,8 @@ import kotlin.reflect.jvm.kotlinFunction
* @author Sebastien Deleuze
* @since 5.2
*/
internal fun <T: Any> deferredToMono(source: Deferred<T>) = GlobalScope.mono { source.await() }
internal fun <T: Any> deferredToMono(source: Deferred<T>) =
GlobalScope.mono(Dispatchers.Unconfined) { source.await() }
/**
* Convert a [Mono] instance to a [Deferred] one.
@ -44,7 +46,8 @@ internal fun <T: Any> deferredToMono(source: Deferred<T>) = GlobalScope.mono { s
* @author Sebastien Deleuze
* @since 5.2
*/
internal fun <T: Any> monoToDeferred(source: Mono<T>) = GlobalScope.async { source.awaitFirstOrNull() }
internal fun <T: Any> monoToDeferred(source: Mono<T>) =
GlobalScope.async(Dispatchers.Unconfined) { source.awaitFirstOrNull() }
/**
* Invoke an handler method converting suspending method to [Mono] if necessary.
@ -55,7 +58,8 @@ internal fun <T: Any> monoToDeferred(source: Mono<T>) = GlobalScope.async { sour
internal fun invokeHandlerMethod(method: Method, bean: Any, vararg args: Any?): Any? {
val function = method.kotlinFunction!!
return if (function.isSuspend) {
GlobalScope.mono { function.callSuspend(bean, *args.sliceArray(0..(args.size-2)))
GlobalScope.mono(Dispatchers.Unconfined) {
function.callSuspend(bean, *args.sliceArray(0..(args.size-2)))
.let { if (it == Unit) null else it} }
.onErrorMap(InvocationTargetException::class) { it.targetException }
}

3
spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt
View File

@ -16,6 +16,7 @@
package org.springframework.web.reactive.function.client
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.reactive.awaitSingle
import kotlinx.coroutines.reactor.mono
@ -77,7 +78,7 @@ suspend fun WebClient.RequestHeadersSpec<out WebClient.RequestHeadersSpec<*>>.aw
* @since 5.2
*/
inline fun <reified T: Any> WebClient.RequestBodySpec.body(crossinline supplier: suspend () -> T)
= body(GlobalScope.mono { supplier.invoke() })
= body(GlobalScope.mono(Dispatchers.Unconfined) { supplier.invoke() })
/**
* Coroutines variant of [WebClient.ResponseSpec.bodyToMono].

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

@ -16,6 +16,7 @@
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
@ -389,7 +390,7 @@ open class CoRouterFunctionDsl(private val init: (CoRouterFunctionDsl.() -> Unit
*/
fun resources(lookupFunction: suspend (ServerRequest) -> Resource?) {
builder.resources {
GlobalScope.mono {
GlobalScope.mono(Dispatchers.Unconfined) {
lookupFunction.invoke(it)
}
}
@ -404,7 +405,7 @@ open class CoRouterFunctionDsl(private val init: (CoRouterFunctionDsl.() -> Unit
}
private fun asHandlerFunction(init: suspend (ServerRequest) -> ServerResponse) = HandlerFunction {
GlobalScope.mono {
GlobalScope.mono(Dispatchers.Unconfined) {
init(it)
}
}

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

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