Open tonny1983 opened 1 year ago
olegz
commented
12 months ago Have you looked at one of our tests with coroutines? - https://github.com/spring-cloud/spring-cloud-function/blob/71ee83ef4b649e673d15139ec726e8237bc60ee8/spring-cloud-function-kotlin/src/test/java/org/springframework/cloud/function/kotlin/ContextFunctionCatalogAutoConfigurationKotlinSuspendTests.java#L34
tonny1983
commented
12 months ago Hi @olegz ,
Thanks a lot for your kind reply.
The test code you mentioned is written in Java, while I use kotest in my project.
When comparing these codes, I got some further information.
1. when "Multi argument Kotlin functions are not currently supported" exception occurs
The exception occurs when using a raw object as input parameter, viz., suspend (Data) -> Unit will cause the exception but suspend (Flow<Data>) -> Unit will not.
The reason of this result is due to isValidKotlinSuspendConsumer method defined in KotlinLambdaToFunctionAutoConfiguration class which does 4 checks:
isTypeRepresentedByClass(functionType, Function2.class) &&
type.length == 3 &&
CoroutinesUtils.isFlowType(type[0]) &&
CoroutinesUtils.isContinuationUnitType(type[1])Obviously, the third one which checks whether the input patameter is a Flow type or not failed. I'm not sure whethere it is an expected behaviour because no ducoments describe using coroutine in Spring Cloud Function, but it is a normal case to use raw object rather than a Flow object as a suspend method's paramter using kotlin coroutine (CoroutineCrudRepository is just a great example).
*2. the problem of using `Flow<>`** In your mentioned Java test code, I rewrote it to kotlin style and added something to test result of the function:
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class KotlinSuspendFunctionJUnitTest {
private lateinit var context: GenericApplicationContext
private lateinit var catalog: FunctionCatalog
@AfterEach
fun close() {
context.close()
}
@Test
fun typeDiscoveryTests() {
create(arrayOf(KotlinSuspendFlowLambdasConfiguration::class.java))
val functionCatalog = context.getBean(FunctionCatalog::class.java)
val kotlinFunction = functionCatalog.lookup<FunctionInvocationWrapper>("kotlinFunction")
Assertions.assertThat(kotlinFunction.isFunction).isTrue()
Assertions.assertThat(kotlinFunction.inputType.typeName)
.isEqualTo("reactor.core.publisher.Flux<java.lang.String>")
Assertions.assertThat(kotlinFunction.outputType.typeName)
.isEqualTo("reactor.core.publisher.Flux<java.lang.String>")
// The following code runs correctly
val result = kotlinFunction.apply(Flux.just("abcd")) as Flux<String>
StepVerifier.create(result).expectNext("ABCD").verifyComplete()
}
private fun create(types: Array<Class<*>>, vararg props: String) {
context = SpringApplicationBuilder(*types).properties(*props).run() as GenericApplicationContext
catalog = context.getBean(FunctionCatalog::class.java)
}
}As you can see, if using Flux which is the assertion checked input and output type, the test passes without errors.
However, if trying to use Flow which is just as the same as the definition of the kotlinFunction method, an exception occured and the test failed:
CASE 1: Flow as input and Flux as output
val result1 = kotlinFunction.apply(flowOf("abcd"))
val result2 = result1 as Flux<String>
StepVerifier.create(result2).expectNext("ABCD").verifyComplete()An exception occurs at the expectNext assertion:
expectation "expectNext(ABCD)" failed (expected: onNext(ABCD); actual: onError(java.lang.ClassCastException: class kotlinx.coroutines.flow.FlowKt__BuildersKt$flowOf$$inlined$unsafeFlow$2 cannot be cast to class java.lang.String (kotlinx.coroutines.flow.FlowKt__BuildersKt$flowOf$$inlined$unsafeFlow$2 is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')))
java.lang.AssertionError: expectation "expectNext(ABCD)" failed (expected: onNext(ABCD); actual: onError(java.lang.ClassCastException: class kotlinx.coroutines.flow.FlowKt__BuildersKt$flowOf$$inlined$unsafeFlow$2 cannot be cast to class java.lang.String (kotlinx.coroutines.flow.FlowKt__BuildersKt$flowOf$$inlined$unsafeFlow$2 is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')))
at reactor.test.MessageFormatter.assertionError(MessageFormatter.java:115)
at reactor.test.MessageFormatter.failPrefix(MessageFormatter.java:104)
at reactor.test.MessageFormatter.fail(MessageFormatter.java:73)
at reactor.test.MessageFormatter.failOptional(MessageFormatter.java:88)
at reactor.test.DefaultStepVerifierBuilder.lambda$addExpectedValue$10(DefaultStepVerifierBuilder.java:509)
at reactor.test.DefaultStepVerifierBuilder$SignalEvent.test(DefaultStepVerifierBuilder.java:2289)
at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onSignal(DefaultStepVerifierBuilder.java:1529)
at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onExpectation(DefaultStepVerifierBuilder.java:1477)
at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onError(DefaultStepVerifierBuilder.java:1129)
at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:134)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:134)
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onError(MonoFlatMapMany.java:255)
at kotlinx.coroutines.reactive.FlowSubscription.flowProcessing(ReactiveFlow.kt:215)
at kotlinx.coroutines.reactive.FlowSubscription.access$flowProcessing(ReactiveFlow.kt:187)
at kotlinx.coroutines.reactive.FlowSubscription$createInitialContinuation$1$1.invoke(ReactiveFlow.kt:204)
at kotlinx.coroutines.reactive.FlowSubscription$createInitialContinuation$1$1.invoke(ReactiveFlow.kt:204)
at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt$createCoroutineUnintercepted$$inlined$createCoroutineFromSuspendFunction$IntrinsicsKt__IntrinsicsJvmKt$2.invokeSuspend(IntrinsicsJvm.kt:270)
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:108)
at kotlinx.coroutines.EventLoop.processUnconfinedEvent(EventLoop.common.kt:68)
at kotlinx.coroutines.internal.DispatchedContinuationKt.resumeCancellableWith(DispatchedContinuation.kt:375)
at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:30)
at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable$default(Cancellable.kt:25)
at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:110)
at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:126)
at kotlinx.coroutines.reactor.MonoKt.monoInternal$lambda$2(Mono.kt:92)
at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:58)
at reactor.core.publisher.Flux.subscribe(Flux.java:8773)
at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.toVerifierAndSubscribe(DefaultStepVerifierBuilder.java:891)
at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.verify(DefaultStepVerifierBuilder.java:831)
at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.verify(DefaultStepVerifierBuilder.java:823)
at reactor.test.DefaultStepVerifierBuilder.verifyComplete(DefaultStepVerifierBuilder.java:690)CASE 2: Flux as input and Flow as output
val result1 = kotlinFunction.apply(Flux.just("abcd"))
val result2 = result1 as Flow<String>An exception occurs on the type cast:
class reactor.core.publisher.FluxMap cannot be cast to class kotlinx.coroutines.flow.Flow (reactor.core.publisher.FluxMap and kotlinx.coroutines.flow.Flow are in unnamed module of loader 'app')
java.lang.ClassCastException: class reactor.core.publisher.FluxMap cannot be cast to class kotlinx.coroutines.flow.Flow (reactor.core.publisher.FluxMap and kotlinx.coroutines.flow.Flow are in unnamed module of loader 'app')
at cc.tonny.springfunctiondemo.KotlinSuspendFunctionKotest$1$1.invokeSuspend(KotlinSuspendFunctionKotest.kt:27)
at cc.tonny.springfunctiondemo.KotlinSuspendFunctionKotest$1$1.invoke(KotlinSuspendFunctionKotest.kt)CASE 3: Flow as input and Flow as output
The result is as the same as CASE 2.
Describe the bug According to the docs, a lambda can be used to implement
java.utl.function.Cusumer.However, if calls a suspend method in the body, the lambda must add
suspendkeywords which casuses an exception ofjava.lang.UnsupportedOperationException: Multi argument Kotlin functions are not currently supported.Sample Suppose I have a
DataRepositoryinterface which extendsCoroutineCrudRepository, hence theDataRepository#savemethod is asuspendmethod (In theCoroutineCrudRepositoryinterface, it issuspend fun <S : T> save(entity: S): T).And the function class is
and it cause the exception
Other tries If using
kotlinx.coroutines.flow.Flow, a non-suspend methodCoroutineCrudRepository#saveAllcan be used. Therefore, the lambda signature can be changed to(Flow<Data>) -> Unitwhich causes no exceptions in startup.However, if call the function with a message, another exception occurs:
ugly workaound Calling
suspendmethod in arunBlockingblock that makes the Consumer itself be not in a suspend one.The version of spring-cloud-function is 4.0.5 (spring-cloud 2022.0.4).