Browse Source
Migrate `CoroutinesUtils` from Kotlin code to Java and drop the `kotlin-coroutines` module. This update removes the need for Kotlin tooling IDE plugins to be installed. Closes gh-27379pull/27415/head
Phillip Webb
3 years ago
committed by
Sam Brannen
11 changed files with 101 additions and 124 deletions
@ -1,35 +0,0 @@
@@ -1,35 +0,0 @@
|
||||
description = "Spring Core Coroutines support" |
||||
|
||||
apply plugin: "kotlin" |
||||
|
||||
configurations { |
||||
classesOnlyElements { |
||||
canBeConsumed = true |
||||
canBeResolved = false |
||||
} |
||||
} |
||||
|
||||
artifacts { |
||||
classesOnlyElements(compileKotlin.destinationDir) |
||||
} |
||||
|
||||
dependencies { |
||||
api("org.jetbrains.kotlin:kotlin-reflect") |
||||
api("org.jetbrains.kotlin:kotlin-stdlib") |
||||
api("io.projectreactor:reactor-core") |
||||
api("org.jetbrains.kotlinx:kotlinx-coroutines-core") |
||||
api("org.jetbrains.kotlinx:kotlinx-coroutines-reactor") |
||||
} |
||||
|
||||
eclipse { |
||||
project { |
||||
buildCommand "org.jetbrains.kotlin.ui.kotlinBuilder" |
||||
buildCommand "org.eclipse.jdt.core.javabuilder" |
||||
natures "org.jetbrains.kotlin.core.kotlinNature" |
||||
natures "org.eclipse.jdt.core.javanature" |
||||
linkedResource name: "kotlin_bin", type: "2", locationUri: "org.jetbrains.kotlin.core.filesystem:/" + project.name + "/kotlin_bin" |
||||
} |
||||
classpath { |
||||
containers "org.jetbrains.kotlin.core.KOTLIN_CONTAINER" |
||||
} |
||||
} |
||||
@ -1,71 +0,0 @@
@@ -1,71 +0,0 @@
|
||||
/* |
||||
* Copyright 2002-2021 the original author or authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* https://www.apache.org/licenses/LICENSE-2.0 |
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
@file:JvmName("CoroutinesUtils") |
||||
package org.springframework.core |
||||
|
||||
import kotlinx.coroutines.* |
||||
import kotlinx.coroutines.flow.Flow |
||||
import kotlinx.coroutines.reactive.awaitSingleOrNull |
||||
import kotlinx.coroutines.reactor.asFlux |
||||
|
||||
import kotlinx.coroutines.reactor.mono |
||||
import org.reactivestreams.Publisher |
||||
import reactor.core.publisher.Mono |
||||
import java.lang.reflect.InvocationTargetException |
||||
import java.lang.reflect.Method |
||||
import kotlin.reflect.full.callSuspend |
||||
import kotlin.reflect.jvm.kotlinFunction |
||||
|
||||
/** |
||||
* Convert a [Deferred] instance to a [Mono] one. |
||||
* |
||||
* @author Sebastien Deleuze |
||||
* @since 5.2 |
||||
*/ |
||||
internal fun <T: Any> deferredToMono(source: Deferred<T>) = |
||||
mono(Dispatchers.Unconfined) { source.await() } |
||||
|
||||
/** |
||||
* Convert a [Mono] instance to a [Deferred] one. |
||||
* |
||||
* @author Sebastien Deleuze |
||||
* @since 5.2 |
||||
*/ |
||||
@Suppress("DEPRECATION") |
||||
@OptIn(DelicateCoroutinesApi::class) |
||||
internal fun <T: Any> monoToDeferred(source: Mono<T>) = |
||||
GlobalScope.async(Dispatchers.Unconfined) { source.awaitSingleOrNull() } |
||||
|
||||
/** |
||||
* Invoke a suspending function and converts it to [Mono] or [reactor.core.publisher.Flux]. |
||||
* |
||||
* @author Sebastien Deleuze |
||||
* @since 5.2 |
||||
*/ |
||||
@Suppress("UNCHECKED_CAST") |
||||
fun invokeSuspendingFunction(method: Method, target: Any, vararg args: Any?): Publisher<*> { |
||||
val function = method.kotlinFunction!! |
||||
val mono = mono(Dispatchers.Unconfined) { |
||||
function.callSuspend(target, *args.sliceArray(0..(args.size-2))).let { if (it == Unit) null else it } |
||||
}.onErrorMap(InvocationTargetException::class.java) { it.targetException } |
||||
return if (function.returnType.classifier == Flow::class) { |
||||
mono.flatMapMany { (it as Flow<Any>).asFlux() } |
||||
} |
||||
else { |
||||
mono |
||||
} |
||||
} |
||||
@ -0,0 +1,98 @@
@@ -0,0 +1,98 @@
|
||||
/* |
||||
* Copyright 2002-2021 the original author or authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.springframework.core; |
||||
|
||||
import java.lang.reflect.InvocationTargetException; |
||||
import java.lang.reflect.Method; |
||||
import java.util.Objects; |
||||
|
||||
import kotlin.Unit; |
||||
import kotlin.jvm.JvmClassMappingKt; |
||||
import kotlin.reflect.KClassifier; |
||||
import kotlin.reflect.KFunction; |
||||
import kotlin.reflect.full.KCallables; |
||||
import kotlin.reflect.jvm.ReflectJvmMapping; |
||||
import kotlinx.coroutines.BuildersKt; |
||||
import kotlinx.coroutines.CoroutineStart; |
||||
import kotlinx.coroutines.Deferred; |
||||
import kotlinx.coroutines.Dispatchers; |
||||
import kotlinx.coroutines.GlobalScope; |
||||
import kotlinx.coroutines.flow.Flow; |
||||
import kotlinx.coroutines.reactor.MonoKt; |
||||
import kotlinx.coroutines.reactor.ReactorFlowKt; |
||||
import org.reactivestreams.Publisher; |
||||
import reactor.core.publisher.Flux; |
||||
import reactor.core.publisher.Mono; |
||||
|
||||
/** |
||||
* Utilities for working with Kotlin Coroutines. |
||||
* |
||||
* @author Sebastien Deleuze |
||||
* @author Phillip Webb |
||||
* @since 5.2 |
||||
*/ |
||||
public final class CoroutinesUtils { |
||||
|
||||
private CoroutinesUtils() { |
||||
} |
||||
|
||||
/** |
||||
* Convert a {@link Deferred} instance to a {@link Mono}. |
||||
*/ |
||||
public static <T> Mono<T> deferredToMono(Deferred<T> source) { |
||||
return MonoKt.mono(Dispatchers.getUnconfined(), |
||||
(scope, continuation) -> source.await(continuation)); |
||||
} |
||||
|
||||
/** |
||||
* Convert a {@link Mono} instance to a {@link Deferred}. |
||||
*/ |
||||
public static <T> Deferred<T> monoToDeferred(Mono<T> source) { |
||||
return BuildersKt.async(GlobalScope.INSTANCE, Dispatchers.getUnconfined(), |
||||
CoroutineStart.DEFAULT, |
||||
(scope, continuation) -> MonoKt.awaitSingleOrNull(source, continuation)); |
||||
} |
||||
|
||||
/** |
||||
* Invoke a suspending function and converts it to {@link Mono} or |
||||
* {@link Flux}. |
||||
*/ |
||||
public static Publisher<?> invokeSuspendingFunction(Method method, Object target, Object... args) { |
||||
KFunction<?> function = Objects.requireNonNull(ReflectJvmMapping.getKotlinFunction(method)); |
||||
KClassifier classifier = function.getReturnType().getClassifier(); |
||||
Mono<Object> mono = MonoKt.mono(Dispatchers.getUnconfined(), (scope, continuation) -> |
||||
KCallables.callSuspend(function, getSuspendedFunctionArgs(target, args), continuation)) |
||||
.filter(result -> !Objects.equals(result, Unit.INSTANCE)) |
||||
.onErrorMap(InvocationTargetException.class, InvocationTargetException::getTargetException); |
||||
if (classifier.equals(JvmClassMappingKt.getKotlinClass(Flow.class))) { |
||||
return mono.flatMapMany(CoroutinesUtils::asFlux); |
||||
} |
||||
return mono; |
||||
} |
||||
|
||||
private static Object[] getSuspendedFunctionArgs(Object target, Object... args) { |
||||
Object[] functionArgs = new Object[args.length]; |
||||
functionArgs[0] = target; |
||||
System.arraycopy(args, 0, functionArgs, 1, args.length - 1); |
||||
return functionArgs; |
||||
} |
||||
|
||||
private static Flux<?> asFlux(Object flow) { |
||||
return ReactorFlowKt.asFlux(((Flow<?>) flow)); |
||||
} |
||||
|
||||
} |
||||
Loading…
Reference in new issue