Netty: ApplicationStarted event is fired before the server starts acc…

…epting connections (#3286)

KTOR-4259 Add ServerReady event, which is fired when an engine is ready to accept connections

ktor-server/ktor-server-cio/jvmAndNix/src/io/ktor/server/cio/CIOApplicationEngine.kt

@@ -4,6 +4,7 @@
 
 package io.ktor.server.cio
 
+import io.ktor.events.*
 import io.ktor.http.cio.*
 import io.ktor.server.application.*
 import io.ktor.server.cio.backend.*
@@ -57,6 +58,7 @@ public class CIOApplicationEngine(
 
         runBlocking {
             startupJob.await()
+            environment.monitor.raiseCatching(ServerReady, environment, environment.log)
 
             if (wait) {
                 serverJob.join()

ktor-server/ktor-server-core/api/ktor-server-core.api

@@ -107,6 +107,7 @@ public final class io/ktor/server/application/DefaultApplicationEventsKt {
 	public static final fun getApplicationStopPreparing ()Lio/ktor/events/EventDefinition;
 	public static final fun getApplicationStopped ()Lio/ktor/events/EventDefinition;
 	public static final fun getApplicationStopping ()Lio/ktor/events/EventDefinition;
+	public static final fun getServerReady ()Lio/ktor/events/EventDefinition;
 }
 
 public class io/ktor/server/application/DuplicateApplicationPluginException : java/lang/Exception {

ktor-server/ktor-server-core/jvmAndNix/src/io/ktor/server/application/DefaultApplicationEvents.kt

@@ -22,6 +22,11 @@ public val ApplicationStarting: EventDefinition<Application> = EventDefinition()
  */
 public val ApplicationStarted: EventDefinition<Application> = EventDefinition()
 
+/**
+ * Fired when the server is ready to accept connections
+ */
+public val ServerReady: EventDefinition<ApplicationEnvironment> = EventDefinition()
+
 /**
  * Event definition for an event that is fired when the application is going to stop
  */

ktor-server/ktor-server-host-common/jvm/src/io/ktor/server/engine/ApplicationEngineEnvironmentReloading.kt

@@ -202,11 +202,7 @@ public class ApplicationEngineEnvironmentReloading(
     }
 
     private fun safeRiseEvent(event: EventDefinition<Application>, application: Application) {
-        try {
-            monitor.raise(event, application)
-        } catch (cause: Throwable) {
-            log.error("One or more of the handlers thrown an exception", cause)
-        }
+        monitor.raiseCatching(event, application)
     }
 
     private fun destroyApplication() {

ktor-server/ktor-server-jetty/jvm/src/io/ktor/server/jetty/JettyApplicationEngineBase.kt

@@ -4,6 +4,7 @@
 
 package io.ktor.server.jetty
 
+import io.ktor.events.*
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import kotlinx.coroutines.*
@@ -54,6 +55,8 @@ public open class JettyApplicationEngineBase(
             .map { it.second.withPort((it.first as ServerConnector).localPort) }
         resolvedConnectors.complete(connectors)
 
+        environment.monitor.raiseCatching(ServerReady, environment, environment.log)
+
         if (wait) {
             server.join()
             stop(1, 5, TimeUnit.SECONDS)

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/NettyApplicationEngine.kt

@@ -4,6 +4,7 @@
 
 package io.ktor.server.netty
 
+import io.ktor.events.*
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import io.ktor.util.network.*
@@ -222,6 +223,8 @@ public class NettyApplicationEngine(
             throw cause
         }
 
+        environment.monitor.raiseCatching(ServerReady, environment, environment.log)
+
         cancellationDeferred = stopServerOnCancellation()
 
         if (wait) {
@@ -310,7 +313,7 @@ public class EventLoopGroupProxy(
         private fun markParkingProhibited() {
             try {
                 prohibitParkingFunction?.invoke(null)
-            } catch (cause: Throwable) {
+            } catch (_: Throwable) {
             }
         }
     }

ktor-server/ktor-server-test-suites/jvm/src/io/ktor/server/testing/suites/ConnectionTestSuite.kt

@@ -4,7 +4,15 @@
 
 package io.ktor.server.testing.suites
 
+import io.ktor.client.*
+import io.ktor.client.engine.cio.*
+import io.ktor.client.request.*
+import io.ktor.http.*
+import io.ktor.server.application.*
 import io.ktor.server.engine.*
+import io.ktor.server.response.*
+import io.ktor.server.routing.*
+import io.ktor.test.dispatcher.*
 import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.Assert.*
@@ -36,4 +44,40 @@ abstract class ConnectionTestSuite(val engine: ApplicationEngineFactory<*, *>) {
         assertFalse(addresses.any { it.port == 0 })
         server.stop(50, 1000)
     }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    @Test
+    fun testServerReadyEvent() = runBlocking {
+        val serverStarted = CompletableDeferred<Unit>()
+        val serverPort = withContext(Dispatchers.IO) { ServerSocket(0).use { it.localPort } }
+        val env = applicationEngineEnvironment {
+            connector { port = serverPort }
+
+            module {
+                routing {
+                    get("/") {
+                        call.respond(HttpStatusCode.OK)
+                    }
+                }
+            }
+        }
+
+        val server = embeddedServer(engine, env)
+
+        server.environment.monitor.subscribe(ServerReady) {
+            serverStarted.complete(Unit)
+        }
+
+        GlobalScope.launch {
+            server.start(true)
+        }
+
+        withTimeout(5000) {
+            serverStarted.join()
+            val response = HttpClient(CIO).get("http://127.0.0.1:$serverPort/")
+            assertEquals(HttpStatusCode.OK, response.status)
+        }
+
+        server.stop(50, 100)
+    }
 }

ktor-server/ktor-server-tomcat/jvm/src/io/ktor/server/tomcat/TomcatApplicationEngine.kt

@@ -4,6 +4,7 @@
 
 package io.ktor.server.tomcat
 
+import io.ktor.events.*
 import io.ktor.http.*
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
@@ -147,6 +148,7 @@ public class TomcatApplicationEngine(
         val connectors = server.service.findConnectors().zip(environment.connectors)
             .map { it.second.withPort(it.first.localPort) }
         resolvedConnectors.complete(connectors)
+        environment.monitor.raiseCatching(ServerReady, environment, environment.log)
 
         cancellationDeferred = stopServerOnCancellation()
         if (wait) {

ktor-shared/ktor-events/api/ktor-events.api

@@ -9,3 +9,8 @@ public final class io/ktor/events/Events {
 	public final fun unsubscribe (Lio/ktor/events/EventDefinition;Lkotlin/jvm/functions/Function1;)V
 }
 
+public final class io/ktor/events/EventsKt {
+	public static final fun raiseCatching (Lio/ktor/events/Events;Lio/ktor/events/EventDefinition;Ljava/lang/Object;Lorg/slf4j/Logger;)V
+	public static synthetic fun raiseCatching$default (Lio/ktor/events/Events;Lio/ktor/events/EventDefinition;Ljava/lang/Object;Lorg/slf4j/Logger;ILjava/lang/Object;)V
+}
+

ktor-shared/ktor-events/common/src/io/ktor/events/Events.kt

@@ -6,6 +6,7 @@ package io.ktor.events
 
 import io.ktor.util.*
 import io.ktor.util.collections.*
+import io.ktor.util.logging.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.internal.*
 
@@ -61,6 +62,17 @@ public class Events {
     }
 }
 
+/**
+ * Raises an event the same way as [Events.raise] but catches an exception and logs it if the [logger] is provided
+ */
+public fun <T> Events.raiseCatching(definition: EventDefinition<T>, value: T, logger: Logger? = null) {
+    try {
+        raise(definition, value)
+    } catch (cause: Throwable) {
+        logger?.error("Some handlers have thrown an exception", cause)
+    }
+}
+
 /**
  * Specifies signature for the event handler
  */