Repository: tipsy/javalin
Pull request #1560 (2022-06-08 13:23:06)

[core] Hot-fix proposal for manual usage of ctx.req.startAsync() 
Associated with #1559

Pros:
* Well... it's still better than some kind of a manual check based on yet another global variable
* Blocks Javalin's handler servlet processing, so in general it's up to the user to close it

Cons: 
* ~It won't call finishResponse(), because AsyncListener behaves differently than regular execution state and it may result in lack of data passed to the final output right before the call, because listener has priority over http thread managed by Jetty. AsyncEvent provides some request/response, maybe we could somehow try to write here? idk~ I think it works like it was before seeing the impl of `AsyncContext#complete()`.
* A few arguments on Discord I still hold against introducing such change

In 5.x it should be removed anyway imo. Also, I've used the old approach in SseHandler, but we should probably still use the valid, <br>* new *, way to do that.

dzikoysk: Speaking of the `finishResponse`, I don't think it could work even before:

![obraz](https://user-images.githubusercontent.com/4235722/172633934-e47c99db-1c4f-4497-b1b0-e72bedf0803f.png)

Calling the `ctx.req.asyncContext.complete()` explicitly changes state of HttpOutput to CLOSED, so there's just no way to write anything after that. It makes sense, but also shows that even in case of the old implementation, Javalin couldn't write response either. 


Patch:
diff --git a/javalin/src/main/java/io/javalin/http/JavalinServletHandler.kt b/javalin/src/main/java/io/javalin/http/JavalinServletHandler.kt
index 5093b7c9..381312e2 100644
--- a/javalin/src/main/java/io/javalin/http/JavalinServletHandler.kt
+++ b/javalin/src/main/java/io/javalin/http/JavalinServletHandler.kt
@@ -82,7 +82,7 @@ class JavalinServletHandler(
         else
             currentTaskFuture = currentTaskFuture
                 .thenAccept { inputStream -> previousResult = inputStream }
-                .thenCompose { executeNextTask() }  // chain next task into current future
+                .thenCompose { executeNextTask() } // chain next task into current future
                 .exceptionally { throwable -> exceptionMapper.handleUnexpectedThrowable(ctx.res, throwable) } // default catch-all for whole scope, might occur when e.g. finishResponse() will fail
     }
 
@@ -93,6 +93,7 @@ class JavalinServletHandler(
             queueNextTaskOrFinish() // each subsequent task for this stage will be queued and skipped
             return completedFuture(previousResult)
         }
+        val wasAsync = ctx.isAsync() // necessary to detect if user called startAsync() manually
         try {
             /** run code added through submitTask in [JavalinServlet]. This mutates [ctx] */
             task.handler(this)
@@ -102,8 +103,14 @@ class JavalinServletHandler(
             exceptionMapper.handle(exception, ctx)
         }
         return ctx.resultReference.getAndSet(Result(previousResult))
+            .let { result ->
+                when {
+                    ctx.isAsync() && !wasAsync -> result.copy(future = CompletableFuture<Void>()) // GH-1560: freeze JavalinServletHandler infinitely, TODO: Remove it in Javalin 5.x
+                    else -> result
+                }
+            }
             .also { result -> if (!ctx.isAsync() && !result.future.isDone) startAsyncAndAddDefaultTimeoutListeners() } // start async context only if the future is not already completed
-            .also { result -> if (ctx.isAsync()) ctx.req.asyncContext.addTimeoutListener { result.future.cancel(true) } }
+            .also { result -> if (ctx.isAsync()) ctx.req.asyncContext.addListener(onTimeout = { result.future.cancel(true) }) }
             .let { result ->
                 result.future
                     .thenAccept { any -> (result.callback?.accept(any) ?: ctx.contextResolver().defaultFutureCallback(ctx, any)) } // callback after future resolves - modifies ctx result, status, etc
@@ -114,12 +121,12 @@ class JavalinServletHandler(
     }
 
     private fun startAsyncAndAddDefaultTimeoutListeners() = ctx.req.startAsync()
-        .addTimeoutListener { // a timeout avoids the pipeline - we need to handle it manually
+        .addListener(onTimeout = { // a timeout avoids the pipeline - we need to handle it manually
             currentTaskFuture.cancel(true) // cancel current task
             ctx.status(500).result("Request timed out") // default error handling
             errorMapper.handle(ctx.status(), ctx) // user defined error handling
             finishResponse() // write response
-        }
+        })
         .also { asyncCtx -> asyncCtx.timeout = config.asyncRequestTimeout }
 
     /** Writes response to the client and frees resources */
@@ -140,11 +147,16 @@ class JavalinServletHandler(
 /** Checks if request is executed asynchronously */
 private fun Context.isAsync(): Boolean = req.isAsyncStarted
 
-private fun AsyncContext.addTimeoutListener(callback: () -> Unit) = this.apply {
+internal fun AsyncContext.addListener(
+    onComplete: (AsyncEvent) -> Unit = {},
+    onError: (AsyncEvent) -> Unit = {},
+    onStartAsync: (AsyncEvent) -> Unit = {},
+    onTimeout: (AsyncEvent) -> Unit = {},
+) : AsyncContext = apply {
     addListener(object : AsyncListener {
-        override fun onComplete(event: AsyncEvent) {}
-        override fun onError(event: AsyncEvent) {}
-        override fun onStartAsync(event: AsyncEvent) {}
-        override fun onTimeout(event: AsyncEvent) = callback() // this is all we care about
+        override fun onComplete(event: AsyncEvent) = onComplete(event)
+        override fun onError(event: AsyncEvent) = onError(event)
+        override fun onStartAsync(event: AsyncEvent) = onStartAsync(event)
+        override fun onTimeout(event: AsyncEvent) = onTimeout(event)
     })
 }
diff --git a/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt b/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
index e0a46f75..011bedf1 100644
--- a/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
+++ b/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
@@ -3,11 +3,9 @@ package io.javalin.http.sse
 import io.javalin.core.util.Header
 import io.javalin.http.Context
 import io.javalin.http.Handler
+import io.javalin.http.addListener
 import java.io.Closeable
-import java.util.concurrent.CompletableFuture
 import java.util.function.Consumer
-import javax.servlet.AsyncEvent
-import javax.servlet.AsyncListener
 
 internal fun interface CloseSseFunction : Closeable
 
@@ -27,19 +25,15 @@ class SseHandler @JvmOverloads constructor(
                 addHeader(Header.X_ACCEL_BUFFERING, "no") // See https://serverfault.com/a/801629
                 flushBuffer()
             }
+
             ctx.req.startAsync(ctx.req, ctx.res)
             ctx.req.asyncContext.timeout = timeout
+            val closeSse = CloseSseFunction { ctx.req.asyncContext.complete() }
 
-            val closeSse = CompletableFuture<Void>()
-                .also { ctx.future(it) { /* do nothing with the future result in callback */ } }
-                .let { CloseSseFunction { it.complete(null) } }
-
-            ctx.req.asyncContext.addListener(object : AsyncListener {
-                override fun onComplete(event: AsyncEvent) {}
-                override fun onStartAsync(event: AsyncEvent) {}
-                override fun onTimeout(event: AsyncEvent) { closeSse.close() }
-                override fun onError(event: AsyncEvent) { closeSse.close() }
-            })
+            ctx.req.asyncContext.addListener(
+                onTimeout = { closeSse.close() },
+                onError = { closeSse.close() }
+            )
 
             clientConsumer.accept(SseClient(closeSse, ctx))
         }


Test patch:
diff --git a/javalin/src/test/java/io/javalin/TestFuture.kt b/javalin/src/test/java/io/javalin/TestFuture.kt
index 91478c56..fd20a2d8 100644
--- a/javalin/src/test/java/io/javalin/TestFuture.kt
+++ b/javalin/src/test/java/io/javalin/TestFuture.kt
@@ -189,11 +189,25 @@ class TestFuture {
             it.contextResolvers { it.defaultFutureCallback = { ctx, _ -> ctx.result("Ignore result") } }
         }
         TestUtil.test(ignoringServer) { app, http ->
-            app.get("/") {  it.future(CompletableFuture.completedFuture("Success")) }
+            app.get("/") { it.future(CompletableFuture.completedFuture("Success")) }
             assertThat(http.get("/").body).isEqualTo("Ignore result")
         }
     }
 
+    @Test
+    fun `should support legacy usage of asyncStart`() = TestUtil.test(impatientServer) { app, http ->
+        app.get("/") { ctx ->
+            ctx.req.startAsync()
+
+            getFuture("response").thenAccept {
+                ctx.res.outputStream.write(it.toByteArray())
+                ctx.req.asyncContext.complete()
+            }
+        }
+
+        assertThat(http.get("/").body).isEqualTo("response")
+    }
+
     private fun getFuture(result: String?, delay: Long = 10): CompletableFuture<String> {
         val future = CompletableFuture<String>()
         Executors.newSingleThreadScheduledExecutor().schedule({
diff --git a/javalin/src/test/java/io/javalin/TestSse.kt b/javalin/src/test/java/io/javalin/TestSse.kt
index a3c1c60b..199c6296 100644
--- a/javalin/src/test/java/io/javalin/TestSse.kt
+++ b/javalin/src/test/java/io/javalin/TestSse.kt
@@ -1,12 +1,10 @@
 package io.javalin
 
 import io.javalin.http.sse.SseClient
-import io.javalin.http.sse.SseHandler
 import io.javalin.testing.SerializableObject
 import io.javalin.testing.TestUtil
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.Timeout
 import java.util.concurrent.CompletableFuture
 
 class TestSse {
@@ -112,7 +110,7 @@ class TestSse {
         app.sse("/sse") {
             it.sendEvent("Sync event")
             CompletableFuture.runAsync {
-                Thread.sleep(10)
+                Thread.sleep(100)
                 it.sendEvent("Async event")
                 it.close()
             }