SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixed a bug where the consumer function could not subscribe to the incoming flow

Browse Source 
Resolves #701
Resolves #702
This commit is contained in:
Tsypov Dmitriy
2021-05-21 22:05:22 +07:00
committed by Oleg Zhurakousky
parent c2d0590ec3
commit dc2076c77a
2 changed files with 27 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java
View File

@@ -167,6 +167,10 @@ public class KotlinLambdaToFunctionAutoConfiguration {
@Override @Override
public void accept(Object input) { public void accept(Object input) {
if (CoroutinesUtils.isValidSuspendingFunction(kotlinLambdaTarget, input)) {
CoroutinesUtils.invokeSuspendingConsumer(kotlinLambdaTarget, input);
return;
}
this.apply(input); this.apply(input);
} }

39
spring-cloud-function-context/src/main/kotlin/org/springframework/cloud/function/context/config/CoroutinesUtils.kt
View File

@@ -34,10 +34,6 @@ import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn
* *
*/ */
fun isValidSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Boolean {
return arg0 is Flux<*> && kotlinLambdaTarget is Function2<*, *, *>
}
fun getSuspendingFunctionArgType(type: Type): Type { fun getSuspendingFunctionArgType(type: Type): Type {
return getFlowTypeArguments(type) return getFlowTypeArguments(type)
} }
@@ -93,11 +89,13 @@ private fun getContinuationTypeArguments(type: Type): Type {
fun invokeSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Flux<Any> { fun invokeSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Flux<Any> {
val function = kotlinLambdaTarget as SuspendFunction val function = kotlinLambdaTarget as SuspendFunction
val flux = arg0 as Flux<Any> val flux = arg0 as Flux<Any>
return fluxSuspendingFlowFunction(flux, function) return mono(Dispatchers.Unconfined) {
} suspendCoroutineUninterceptedOrReturn<Flow<Any>> {
function.invoke(flux.asFlow(), it)
fun isValidSuspendingSupplier(kotlinLambdaTarget: Any): Boolean { }
return kotlinLambdaTarget is Function1<*, *> }.flatMapMany {
it.asFlux()
}
} }
fun invokeSuspendingSupplier(kotlinLambdaTarget: Any): Flux<Any> { fun invokeSuspendingSupplier(kotlinLambdaTarget: Any): Flux<Any> {
@@ -111,15 +109,24 @@ fun invokeSuspendingSupplier(kotlinLambdaTarget: Any): Flux<Any> {
} }
} }
fun fluxSuspendingFlowFunction(flux: Flux<Any>, target: SuspendFunction): Flux<Any> { fun invokeSuspendingConsumer(kotlinLambdaTarget: Any, arg0: Any) {
return mono(Dispatchers.Unconfined) { val consumer = kotlinLambdaTarget as SuspendConsumer
suspendCoroutineUninterceptedOrReturn<Flow<Any>> { val flux = arg0 as Flux<Any>
target.invoke(flux.asFlow(), it) mono(Dispatchers.Unconfined) {
suspendCoroutineUninterceptedOrReturn<Unit> {
consumer.invoke(flux.asFlow(), it)
} }
}.flatMapMany { }.subscribe()
it.asFlux() }
}
fun isValidSuspendingFunction(kotlinLambdaTarget: Any, arg0: Any): Boolean {
return arg0 is Flux<*> && kotlinLambdaTarget is Function2<*, *, *>
}
fun isValidSuspendingSupplier(kotlinLambdaTarget: Any): Boolean {
return kotlinLambdaTarget is Function1<*, *>
} }
private typealias SuspendFunction = (Any?, Any?) -> Any? private typealias SuspendFunction = (Any?, Any?) -> Any?
private typealias SuspendConsumer = (Any?, Any?) -> Unit?
private typealias SuspendSupplier = (Any?) -> Any? private typealias SuspendSupplier = (Any?) -> Any?