Ensure Flow return type works with Request Scope in RESTEasy Reactive

Fixes: quarkusio#18993

extensions/resteasy-reactive/quarkus-resteasy-reactive-kotlin/runtime/pom.xml

@@ -27,8 +27,8 @@
             <artifactId>kotlinx-coroutines-jdk8</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.jetbrains.kotlinx</groupId>
-            <artifactId>kotlinx-coroutines-reactive</artifactId>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>mutiny-kotlin</artifactId>
         </dependency>
     </dependencies>
 

extensions/resteasy-reactive/quarkus-resteasy-reactive-kotlin/runtime/src/main/kotlin/org/jboss/resteasy/reactive/server/runtime/kotlin/FlowToPublisherHandler.kt

@@ -1,17 +1,34 @@
 package org.jboss.resteasy.reactive.server.runtime.kotlin
 
+import io.smallrye.mutiny.coroutines.asMulti
+import io.vertx.core.Vertx
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.reactive.asPublisher
+import kotlinx.coroutines.launch
 import org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext
 import org.jboss.resteasy.reactive.server.spi.ServerRestHandler
+import javax.enterprise.inject.spi.CDI
 
 class FlowToPublisherHandler : ServerRestHandler {
 
+    private val originalTCCL: ClassLoader = Thread.currentThread().contextClassLoader
+
     override fun handle(requestContext: ResteasyReactiveRequestContext?) {
         val result = requestContext!!.result
         if (result is Flow<*>) {
-            requestContext.result = (result as Flow<Any>) // cast needed for extension function
-                                        .asPublisher()
+
+            val requestScope = requestContext.captureCDIRequestScope()
+            val dispatcher: CoroutineDispatcher = Vertx.currentContext()?.let {VertxDispatcher(it,requestScope)}
+                    ?: throw IllegalStateException("No Vertx context found")
+
+            val coroutineScope = CDI.current().select(ApplicationCoroutineScope::class.java)
+            requestContext.suspend()
+            coroutineScope.get().launch(context = dispatcher) {
+                // ensure the proper CL is not lost in dev-mode
+                Thread.currentThread().contextClassLoader = originalTCCL
+                requestContext.result = result.asMulti()
+                requestContext.resume()
+            }
         }
     }
 }

integration-tests/resteasy-reactive-kotlin/standard/src/main/kotlin/io/quarkus/it/resteasy/reactive/kotlin/FlowResource.kt

@@ -9,16 +9,16 @@ import javax.ws.rs.Produces
 import javax.ws.rs.core.MediaType
 
 @Path("flow")
-class FlowResource {
+class FlowResource(private val uppercaseService: UppercaseService) {
 
     @GET
     @Path("str")
     @Produces(MediaType.SERVER_SENT_EVENTS)
     fun sseStrings() = flow {
-        emit("Hello")
-        emit("From")
-        emit("Kotlin")
-        emit("Flow")
+        emit(uppercaseService.convert("Hello"))
+        emit(uppercaseService.convert("From"))
+        emit(uppercaseService.convert("Kotlin"))
+        emit(uppercaseService.convert("Flow"))
     }
 
     @GET

integration-tests/resteasy-reactive-kotlin/standard/src/main/kotlin/io/quarkus/it/resteasy/reactive/kotlin/UppercaseService.kt

@@ -0,0 +1,10 @@
+package io.quarkus.it.resteasy.reactive.kotlin
+
+import java.util.Locale
+import javax.enterprise.context.RequestScoped
+
+@RequestScoped
+class UppercaseService {
+
+    fun convert(input: String) = input.uppercase(Locale.ROOT)
+}

integration-tests/resteasy-reactive-kotlin/standard/src/test/kotlin/io/quarkus/it/resteasy/reactive/kotlin/FlowResourceTest.kt

@@ -20,7 +20,7 @@ class FlowResourceTest {
     @Test
     fun testSeeStrings() {
         testSse("str", 5) {
-            assertThat(it).containsExactly("Hello", "From", "Kotlin", "Flow")
+            assertThat(it).containsExactly("HELLO", "FROM", "KOTLIN", "FLOW")
         }
     }