feat: suspend handler (#11)

Author: MukjepScarlet

build.gradle.kts

@@ -11,7 +11,7 @@ val authorName = "ccbluex"
 val projectUrl = "https://github.com/ccbluex/netty-httpserver"
 
 group = "net.ccbluex"
-version = "2.4.2"
+version = "2.4.3-alpha.1"
 
 repositories {
     mavenCentral()
@@ -31,8 +31,10 @@ dependencies {
     api(libs.bundles.netty)
     api(libs.gson)
     api(libs.tika.core)
+    api(libs.coroutines.core)
 
     testImplementation(kotlin("test"))
+    testImplementation(libs.coroutines.test)
     testImplementation("com.squareup.retrofit2:retrofit:2.9.0")
     testImplementation("com.squareup.retrofit2:converter-gson:2.9.0")
 }

src/main/kotlin/net/ccbluex/netty/http/HttpConductor.kt

@@ -34,7 +34,7 @@ import net.ccbluex.netty.http.util.httpNoContent
  * @param context The request context to process.
  * @return The response to the request.
  */
-internal fun HttpServer.processRequestContext(context: RequestContext) = runCatching {
+internal suspend fun HttpServer.processRequestContext(context: RequestContext) = runCatching {
     val content = context.contentBuffer.toByteArray()
     val method = context.httpMethod
 

src/main/kotlin/net/ccbluex/netty/http/HttpServerHandler.kt

@@ -26,9 +26,16 @@ import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpRequest
 import io.netty.handler.codec.http.LastHttpContent
 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory
+import kotlinx.coroutines.CoroutineExceptionHandler
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
 import net.ccbluex.netty.http.HttpServer.Companion.logger
 import net.ccbluex.netty.http.middleware.Middleware
 import net.ccbluex.netty.http.model.RequestContext
+import net.ccbluex.netty.http.util.forEachIsInstance
 import net.ccbluex.netty.http.websocket.WebSocketHandler
 import java.net.URLDecoder
 
@@ -40,13 +47,37 @@ import java.net.URLDecoder
 internal class HttpServerHandler(private val server: HttpServer) : ChannelInboundHandlerAdapter() {
 
     private val localRequestContext = ThreadLocal<RequestContext>()
+    private lateinit var channelScope: CoroutineScope
 
     /**
      * Extension property to get the WebSocket URL from an HttpRequest.
      */
     private val HttpRequest.webSocketUrl: String
         get() = "ws://${headers().get("Host")}${uri()}"
 
+    /**
+     * Adds the [CoroutineScope] of current [io.netty.channel.Channel].
+     */
+    override fun handlerAdded(ctx: ChannelHandlerContext) {
+        super.handlerAdded(ctx)
+
+        val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
+            val ctxName = ctx.name()
+            val channelId = ctx.channel().id().asLongText()
+            logger.error(
+                "Uncaught coroutine error in [ctx: $ctxName, channel: $channelId]",
+                throwable
+            )
+        }
+
+        channelScope = CoroutineScope(
+            ctx.channel().eventLoop().asCoroutineDispatcher()
+                    + CoroutineName("${ctx.name()}#${ctx.channel().id().asShortText()}")
+                    + exceptionHandler
+        )
+        ctx.channel().closeFuture().addListener { channelScope.cancel() }
+    }
+
     /**
      * Reads the incoming messages and processes HTTP requests.
      *
@@ -68,11 +99,11 @@ internal class HttpServerHandler(private val server: HttpServer) : ChannelInboun
                 if (connection.equals("Upgrade", ignoreCase = true) &&
                     upgrade.equals("WebSocket", ignoreCase = true)) {
 
-                    server.middlewares.filterIsInstance<Middleware.OnWebSocketUpgrade>().forEach { middleware ->
+                    server.middlewares.forEachIsInstance<Middleware.OnWebSocketUpgrade> { middleware ->
                         val response = middleware.invoke(ctx, msg)
                         if (response != null) {
                             ctx.writeAndFlush(response)
-                            return
+                            return super.channelRead(ctx, msg)
                         }
                     }
 
@@ -99,15 +130,15 @@ internal class HttpServerHandler(private val server: HttpServer) : ChannelInboun
                         URLDecoder.decode(msg.uri(), Charsets.UTF_8),
                         msg.headers(),
                     )
-                    
+
                     localRequestContext.set(requestContext)
                 }
             }
 
             is HttpContent -> {
                 val requestContext = localRequestContext.get() ?: run {
                     logger.warn("Received HttpContent without HttpRequest")
-                    return
+                    return super.channelRead(ctx, msg)
                 }
 
                 // Append content to the buffer
@@ -117,18 +148,21 @@ internal class HttpServerHandler(private val server: HttpServer) : ChannelInboun
                 if (msg is LastHttpContent) {
                     localRequestContext.remove()
 
-                    server.middlewares.filterIsInstance<Middleware.OnRequest>().forEach { middleware ->
+                    server.middlewares.forEachIsInstance<Middleware.OnRequest> { middleware ->
                         val response = middleware.invoke(requestContext)
                         if (response != null) {
                             ctx.writeAndFlush(response)
-                            return
+                            return super.channelRead(ctx, msg)
                         }
                     }
-                    var response = server.processRequestContext(requestContext)
-                    server.middlewares.filterIsInstance<Middleware.OnResponse>().forEach { middleware ->
-                        response = middleware.invoke(requestContext, response)
+
+                    channelScope.launch {
+                        var response = server.processRequestContext(requestContext)
+                        server.middlewares.forEachIsInstance<Middleware.OnResponse> { middleware ->
+                            response = middleware.invoke(requestContext, response)
+                        }
+                        ctx.writeAndFlush(response)
                     }
-                    ctx.writeAndFlush(response)
                 }
             }
 

src/main/kotlin/net/ccbluex/netty/http/coroutines/NettyCoroutine.kt

@@ -0,0 +1,41 @@
+package net.ccbluex.netty.http.coroutines
+
+import io.netty.util.concurrent.Future
+import io.netty.util.concurrent.GenericFutureListener
+import kotlinx.coroutines.CancellableContinuation
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.suspendCancellableCoroutine
+
+/**
+ * Suspend until this Netty Future completes.
+ *
+ * Returns the Future result. Throws on failure or cancellation.
+ */
+suspend fun <V, F : Future<V>> F.suspend(): V {
+    if (isDone) return unwrapDone().getOrThrow()
+
+    return suspendCancellableCoroutine { cont ->
+        addListener(futureContinuationListener(cont))
+
+        cont.invokeOnCancellation {
+            this.cancel(false)
+        }
+    }
+}
+
+private fun <V, F : Future<V>> futureContinuationListener(
+    cont: CancellableContinuation<V>
+): GenericFutureListener<F> = GenericFutureListener { future ->
+    if (cont.isActive) {
+        cont.resumeWith(future.unwrapDone())
+    }
+}
+
+private fun <V, F : Future<V>> F.unwrapDone(): Result<V> =
+    when {
+        isSuccess -> Result.success(this.now)
+        isCancelled -> Result.failure(CancellationException("Netty Future was cancelled"))
+        else -> Result.failure(
+            this.cause() ?: IllegalStateException("Future failed without cause")
+        )
+    }

src/main/kotlin/net/ccbluex/netty/http/model/RequestHandler.kt

@@ -3,5 +3,5 @@ package net.ccbluex.netty.http.model
 import io.netty.handler.codec.http.FullHttpResponse
 
 fun interface RequestHandler {
-    fun handle(request: RequestObject): FullHttpResponse
+    suspend fun handle(request: RequestObject): FullHttpResponse
 }

src/main/kotlin/net/ccbluex/netty/http/rest/FileServant.kt

@@ -37,7 +37,7 @@ class FileServant(part: String, private val baseFolder: File) : Node(part) {
 
     override val isExecutable = true
 
-    override fun handle(request: RequestObject): FullHttpResponse {
+    override suspend fun handle(request: RequestObject): FullHttpResponse {
         val path = request.remainingPath
         val sanitizedPath = path.replace("..", "")
         val file = baseFolder.resolve(sanitizedPath)

src/main/kotlin/net/ccbluex/netty/http/rest/Node.kt

@@ -135,7 +135,7 @@ open class Node(val part: String) : RequestHandler {
      * @param request The request object.
      * @return The HTTP response.
      */
-    override fun handle(request: RequestObject): FullHttpResponse = throw NotImplementedError()
+    override suspend fun handle(request: RequestObject): FullHttpResponse = throw NotImplementedError()
 
     /**
      * Checks if the node matches a part of the path and HTTP method.

src/main/kotlin/net/ccbluex/netty/http/rest/Route.kt

@@ -33,7 +33,7 @@ import net.ccbluex.netty.http.model.RequestObject
 open class Route(name: String, private val method: HttpMethod, val handler: RequestHandler)
     : Node(name) {
     override val isExecutable = true
-    override fun handle(request: RequestObject) = handler.handle(request)
+    override suspend fun handle(request: RequestObject) = handler.handle(request)
     override fun matchesMethod(method: HttpMethod) =
         this.method == method && super.matchesMethod(method)
 

src/main/kotlin/net/ccbluex/netty/http/rest/ZipServant.kt

@@ -110,7 +110,7 @@ class ZipServant(part: String, zipInputStream: InputStream) : Node(part) {
         return files
     }
 
-    override fun handle(request: RequestObject): FullHttpResponse {
+    override suspend fun handle(request: RequestObject): FullHttpResponse {
         val path = request.remainingPath.removePrefix("/")
         val cleanPath = path.substringBefore("?")
         val sanitizedPath = cleanPath.replace("..", "")

src/main/kotlin/net/ccbluex/netty/http/util/IterableUtils.kt

@@ -0,0 +1,9 @@
+package net.ccbluex.netty.http.util
+
+inline fun <reified E> Iterable<*>.forEachIsInstance(action: (E) -> Unit) {
+    for (it in this) {
+        if (it is E) {
+            action(it)
+        }
+    }
+}