Support Kotlin suspend functions when using @ConsumeEvent

[u6f6o]

Describe the bug

I added a method that should consume an event published by the vertx event bus:

@ConsumeEvent("events.double-opt-in")
suspend fun consume(event: DoubleOptInRequest) {
    Log.info("Received internal event: $event")
    vertxEventBus.send("events.processed", event)
}

When I build the application, the build fails with the following exception:

Execution failed for task ':quarkusBuild'.
> io.quarkus.builder.BuildException: Build failure: Build failed due to errors
    [error]: Build step io.quarkus.vertx.deployment.VertxProcessor#collectEventConsumers threw an exception: java.lang.IllegalStateException: An event consumer business method must accept exactly one parameter: [emailer.events.payload.TrackableEvent, kotlin.coroutines.Continuation<? super kotlin.Unit>] [method: java.lang.Object consume(emailer.events.payload.TrackableEvent event, kotlin.coroutines.Continuation<? super kotlin.Unit> $completion), bean:CLASS bean [types=[emailer.events.EventHandler, java.lang.Object], qualifiers=[@Default, @Any], target=emailer.events.EventHandler]]
    at io.quarkus.vertx.deployment.VertxProcessor.collectEventConsumers(VertxProcessor.java:118)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at io.quarkus.deployment.ExtensionLoader$3.execute(ExtensionLoader.java:944)
    at io.quarkus.builder.BuildContext.run(BuildContext.java:277)
    at org.jboss.threads.ContextHandler$1.runWith(ContextHandler.java:18)
    at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2449)
    at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1478)
    at java.base/java.lang.Thread.run(Thread.java:829)
    at org.jboss.threads.JBossThread.run(JBossThread.java:501)

Looks like, suspend functions with this annotation do not work properly. If I remove the suspend keyword everything works as expected.

Expected behavior

It should be possible to receive events using kotlin suspend functions.

Actual behavior

Using suspend functions with the ConsumeEvent annotation results in a failing build

How to Reproduce?

No response

Output of uname -a or ver

Darwin MB-07-P15.fritz.box 19.6.0 Darwin Kernel Version 19.6.0: Mon Aug 31 22:12:52 PDT 2020; root:xnu-6153.141.2~1/RELEASE_X86_64 x86_64

Output of java -version

openjdk version "17.0.3" 2022-04-19 OpenJDK Runtime Environment Homebrew (build 17.0.3+0) OpenJDK 64-Bit Server VM Homebrew (build 17.0.3+0, mixed mode, sharing)

GraalVM version (if different from Java)

No response

Quarkus version or git rev

2.10.2.Final

Build tool (ie. output of mvnw --version or gradlew --version)

------------------------------------------------------------ Gradle 7.4.2 ------------------------------------------------------------ Build time: 2022-03-31 15:25:29 UTC Revision: 540473b8118064efcc264694cbcaa4b677f61041 Kotlin: 1.5.31 Groovy: 3.0.9 Ant: Apache Ant(TM) version 1.10.11 compiled on July 10 2021 JVM: 17.0.3 (Homebrew 17.0.3+0) OS: Mac OS X 10.15.7 x86_64

Additional information

No response

[quarkus-bot]

/cc @evanchooly

[geoand]

Indeed @ConsumeEvent does not work with Kotin suspend functions currently.

[tpodg]

Are there any plans to support this?

[geoand]

It's pretty low on the list of priorities

[Ladicek]

I have an experimental branch where I rewrote various things in Quarkus that generate their own invoker classes to use CDI 4.1 method invokers. With them, supporting Kotlin suspend functions is straightforward, just a matter of configuring an invoker wrapper, which is one simple Kotlin class. One of the things I rewrote is also @ConsumeEvent, and when we bump to CDI 4.1, I plan to submit PRs with those rewrites. This might come almost for free then.

[popaaaandrei]

You can use this setup:

/// scope for the suspending functions
private val coroutineScope = CoroutineScope(vertx.dispatcher())

You would want to execute the coroutines on the Vertx dispatcher.

@OptIn(ExperimentalCoroutinesApi::class)
@ConsumeEvent()
fun process(event: Event): Uni<Unit> {
    // returns a completed Uni completed when the processing is finished
    val deferred = coroutineScope.async {
         /// call suspending function here
    }

    // map to Uni
    return deferred.asUni()
}

You'll need these 2:

implementation("io.vertx:vertx-lang-kotlin-coroutines")
implementation("io.smallrye.reactive:mutiny-kotlin")

[mschorsch]

Thanks you @Ladicek