Update Ktor integration to Ktor 2.x (#391)

The Ktor 2.x version made some breaking changes, this updates the
ktor-core module to use the new interfaces.

This also bumps up the Kotlin version to a compatible version.

Author: tlusk

build.gradle.kts

@@ -8,7 +8,7 @@ plugins {
     `java-library`
     checkstyle
     jacoco
-    kotlin("jvm") version "1.4.32"
+    kotlin("jvm") version "1.6.21"
     id("com.github.spotbugs")
     id("com.jashmore.gradle.github.release")
     id("org.jlleitschuh.gradle.ktlint") apply false

examples/ktor-example/src/main/kotlin/com/jashmore/sqs/examples/KtorApplicationExample.kt

@@ -7,16 +7,16 @@ import com.jashmore.sqs.elasticmq.ElasticMqSqsAsyncClient
 import com.jashmore.sqs.ktor.container.batchingMessageListener
 import com.jashmore.sqs.ktor.container.messageListener
 import com.jashmore.sqs.ktor.container.prefetchingMessageListener
-import io.ktor.application.call
-import io.ktor.application.log
 import io.ktor.http.ContentType
 import io.ktor.http.HttpStatusCode
-import io.ktor.response.respond
-import io.ktor.response.respondText
-import io.ktor.routing.get
-import io.ktor.routing.routing
+import io.ktor.server.application.call
+import io.ktor.server.application.log
 import io.ktor.server.engine.embeddedServer
 import io.ktor.server.netty.Netty
+import io.ktor.server.response.respond
+import io.ktor.server.response.respondText
+import io.ktor.server.routing.get
+import io.ktor.server.routing.routing
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.future.await
@@ -141,13 +141,7 @@ fun main() {
             }
         }
     }
-    server.start()
-    Runtime.getRuntime().addShutdownHook(
-        Thread {
-            server.stop(1, 30_000)
-        }
-    )
-    Thread.currentThread().join()
+    server.start(wait = true)
 }
 
 suspend fun something(@Suppress("UNUSED_PARAMETER") message: Message) = GlobalScope.launch {

gradle.properties

@@ -11,7 +11,7 @@ guavaVersion=29.0-jre
 immutablesVersion=2.8.9-ea-1
 jacksonVersion=2.12.7
 junitJupiterVersion=5.7.2
-ktorVersion=1.6.0
+ktorVersion=2.2.3
 logbackVersion=1.2.11
 lombokVersion=1.18.26
 mockitoVersion=3.11.2

ktor/core/src/main/kotlin/com/jashmore/sqs/ktor/container/KtorCoreExtension.kt

@@ -6,14 +6,14 @@ import com.jashmore.sqs.core.kotlin.dsl.container.CoreMessageListenerContainerDs
 import com.jashmore.sqs.core.kotlin.dsl.container.FifoMessageListenerContainerDslBuilder
 import com.jashmore.sqs.core.kotlin.dsl.container.PrefetchingMessageListenerContainerDslBuilder
 import com.jashmore.sqs.core.kotlin.dsl.container.coreMessageListener
-import io.ktor.application.Application
-import io.ktor.application.ApplicationEnvironment
-import io.ktor.application.ApplicationStarted
-import io.ktor.application.ApplicationStopped
-import io.ktor.util.pipeline.ContextDsl
+import io.ktor.server.application.Application
+import io.ktor.server.application.ApplicationEnvironment
+import io.ktor.server.application.ApplicationStarted
+import io.ktor.server.application.ApplicationStopped
+import io.ktor.util.KtorDsl
 import software.amazon.awssdk.services.sqs.SqsAsyncClient
 
-@ContextDsl
+@KtorDsl
 fun Application.messageListener(
     identifier: String,
     sqsAsyncClient: SqsAsyncClient,
@@ -23,7 +23,7 @@ fun Application.messageListener(
     return initMessageListener(environment, coreMessageListener(identifier, sqsAsyncClient, queueUrl, config))
 }
 
-@ContextDsl
+@KtorDsl
 fun Application.prefetchingMessageListener(
     identifier: String,
     sqsAsyncClient: SqsAsyncClient,
@@ -33,7 +33,7 @@ fun Application.prefetchingMessageListener(
     return initMessageListener(environment, com.jashmore.sqs.core.kotlin.dsl.container.prefetchingMessageListener(identifier, sqsAsyncClient, queueUrl, init))
 }
 
-@ContextDsl
+@KtorDsl
 fun Application.batchingMessageListener(
     identifier: String,
     sqsAsyncClient: SqsAsyncClient,
@@ -43,7 +43,7 @@ fun Application.batchingMessageListener(
     return initMessageListener(environment, com.jashmore.sqs.core.kotlin.dsl.container.batchingMessageListener(identifier, sqsAsyncClient, queueUrl, init))
 }
 
-@ContextDsl
+@KtorDsl
 fun Application.fifoMessageListener(
     identifier: String,
     sqsAsyncClient: SqsAsyncClient,
@@ -57,13 +57,19 @@ fun initMessageListener(
     environment: ApplicationEnvironment,
     container: MessageListenerContainer
 ): MessageListenerContainer {
+    val hook = Thread { container.stop() }
     environment.monitor.subscribe(ApplicationStarted) {
+        environment.monitor.subscribe(ApplicationStopped) {
+            try {
+                Runtime.getRuntime().removeShutdownHook(hook)
+            } catch (alreadyShuttingDown: IllegalStateException) {
+                // ignore
+            }
+            container.stop()
+        }
+        Runtime.getRuntime().addShutdownHook(hook)
         container.start()
     }
 
-    environment.monitor.subscribe(ApplicationStopped) {
-        container.stop()
-    }
-
     return container
 }

ktor/core/src/test/kotlin/com/jashmore/sqs/ktor/container/KtorCoreExtensionTest.kt

@@ -2,10 +2,8 @@ package com.jashmore.sqs.ktor.container
 
 import com.jashmore.sqs.elasticmq.ElasticMqSqsAsyncClient
 import com.jashmore.sqs.util.LocalSqsAsyncClient
-import io.ktor.application.log
-import io.ktor.server.engine.embeddedServer
-import io.ktor.server.netty.Netty
-import io.ktor.server.testing.withTestApplication
+import io.ktor.server.application.log
+import io.ktor.server.testing.testApplication
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterAll
 import org.junit.jupiter.api.Test
@@ -25,8 +23,8 @@ class KtorCoreExtensionTest {
     fun `message listener can be registered`() {
         val queueUrl = sqsClient.createRandomQueue().get().queueUrl()
         val countDownLatch = CountDownLatch(1)
-        withTestApplication({
-            val server = embeddedServer(Netty, 8080) {
+        testApplication {
+            application {
                 messageListener("core-listener", sqsClient, queueUrl) {
                     processor = lambdaProcessor {
                         method { message ->
@@ -48,8 +46,8 @@ class KtorCoreExtensionTest {
                     }
                 }
             }
-            server.start()
-        }) {
+            startApplication()
+
             sqsClient.sendMessage { it.queueUrl(queueUrl).messageBody("body") }.get()
 
             assertThat(countDownLatch.await(5, TimeUnit.SECONDS)).isTrue()
@@ -60,8 +58,8 @@ class KtorCoreExtensionTest {
     fun `prefetching message listener can be registered`() {
         val queueUrl = sqsClient.createRandomQueue().get().queueUrl()
         val countDownLatch = CountDownLatch(1)
-        withTestApplication({
-            val server = embeddedServer(Netty, 8080) {
+        testApplication {
+            application {
                 prefetchingMessageListener("prefetching-listener", sqsClient, queueUrl) {
                     concurrencyLevel = { 5 }
                     desiredPrefetchedMessages = 1
@@ -74,8 +72,8 @@ class KtorCoreExtensionTest {
                     }
                 }
             }
-            server.start()
-        }) {
+            startApplication()
+
             sqsClient.sendMessage { it.queueUrl(queueUrl).messageBody("body") }.get()
 
             assertThat(countDownLatch.await(5, TimeUnit.SECONDS)).isTrue()
@@ -86,8 +84,8 @@ class KtorCoreExtensionTest {
     fun `batching message listener can be registered`() {
         val queueUrl = sqsClient.createRandomQueue().get().queueUrl()
         val countDownLatch = CountDownLatch(1)
-        withTestApplication({
-            val server = embeddedServer(Netty, 8080) {
+        testApplication {
+            application {
                 batchingMessageListener("batching-listener", sqsClient, queueUrl) {
                     concurrencyLevel = { 5 }
                     processor = lambdaProcessor {
@@ -98,8 +96,8 @@ class KtorCoreExtensionTest {
                     }
                 }
             }
-            server.start()
-        }) {
+            startApplication()
+
             sqsClient.sendMessage { it.queueUrl(queueUrl).messageBody("body") }.get()
 
             assertThat(countDownLatch.await(5, TimeUnit.SECONDS)).isTrue()
@@ -110,8 +108,8 @@ class KtorCoreExtensionTest {
     fun `fifo message listener can be registered`() {
         val queueUrl = sqsClient.createRandomFifoQueue().get().response.queueUrl()
         val countDownLatch = CountDownLatch(1)
-        withTestApplication({
-            val server = embeddedServer(Netty, 8080) {
+        testApplication {
+            application {
                 fifoMessageListener("fifo-listener", sqsClient, queueUrl) {
                     concurrencyLevel = { 5 }
                     processor = lambdaProcessor {
@@ -122,8 +120,8 @@ class KtorCoreExtensionTest {
                     }
                 }
             }
-            server.start()
-        }) {
+            startApplication()
+
             sqsClient.sendMessage {
                 it.queueUrl(queueUrl)
                     .messageBody("body")