migrate to ktor 3.0.0

gradle/libs.versions.toml

@@ -8,7 +8,7 @@ javax-activation = "1.1.1"
 
 korlibs = "5.4.0"
 uuid = "0.8.4"
-ktor = "2.3.12"
+ktor = "3.0.0"
 
 ksp = "2.0.21-1.0.25"
 kotlin-poet = "1.18.1"

tgbotapi.core/src/commonMain/kotlin/dev/inmo/tgbotapi/bot/ktor/base/DownloadFileChannelRequestCallFactory.kt

@@ -8,9 +8,7 @@ import dev.inmo.tgbotapi.utils.ByteReadChannelAllocator
 import dev.inmo.tgbotapi.utils.RiskFeature
 import dev.inmo.tgbotapi.utils.TelegramAPIUrlsKeeper
 import io.ktor.client.HttpClient
-import io.ktor.client.call.receive
 import io.ktor.client.request.get
-import io.ktor.client.statement.HttpStatement
 import io.ktor.client.statement.bodyAsChannel
 import io.ktor.utils.io.*
 import kotlinx.coroutines.*

tgbotapi.core/src/commonMain/kotlin/dev/inmo/tgbotapi/bot/ktor/base/DownloadFileRequestCallFactory.kt

@@ -6,9 +6,9 @@ import dev.inmo.tgbotapi.requests.DownloadFile
 import dev.inmo.tgbotapi.requests.abstracts.Request
 import dev.inmo.tgbotapi.utils.RiskFeature
 import dev.inmo.tgbotapi.utils.TelegramAPIUrlsKeeper
-import io.ktor.client.HttpClient
+import io.ktor.client.*
-import io.ktor.client.request.get
+import io.ktor.client.request.*
-import io.ktor.client.statement.readBytes
+import io.ktor.client.statement.*
 import kotlinx.serialization.json.Json
 
 @RiskFeature
@@ -17,13 +17,13 @@ object DownloadFileRequestCallFactory : KtorCallFactory {
         client: HttpClient,
         urlsKeeper: TelegramAPIUrlsKeeper,
         request: Request<T>,
-        jsonFormatter: Json
+        jsonFormatter: Json,
-    ): T? = (request as? DownloadFile) ?.let {
+    ): T? = (request as? DownloadFile)?.let {
         val fullUrl = urlsKeeper.createFileLinkUrl(it.filePath)
 
         safely {
             @Suppress("UNCHECKED_CAST")
-            client.get(fullUrl).readBytes() as T // always ByteArray
+            client.get(fullUrl).readRawBytes() as T // always ByteArray
         }
     }
 }

tgbotapi.core/src/commonMain/kotlin/dev/inmo/tgbotapi/bot/ktor/base/MultipleClientKtorRequestsExecutor.kt

@@ -38,7 +38,7 @@ internal expect inline fun platformClientCopy(client: HttpClient): HttpClient
  * @param requestExecutorsCount Amount of [DefaultKtorRequestsExecutor] which will be created and used under the
  * hood
  */
-class MultipleClientKtorRequestsExecutor (
+class MultipleClientKtorRequestsExecutor(
     telegramAPIUrlsKeeper: TelegramAPIUrlsKeeper,
     callsFactories: List<KtorCallFactory>,
     excludeDefaultFactories: Boolean,
@@ -47,7 +47,7 @@ class MultipleClientKtorRequestsExecutor (
     pipelineStepsHolder: TelegramBotPipelinesHandler,
     requestExecutorsCount: Int,
     logger: KSLog,
-    clientFactory: () -> HttpClient
+    clientFactory: () -> HttpClient,
 ) : BaseRequestsExecutor(telegramAPIUrlsKeeper) {
     private val requestExecutors = (0 until requestExecutorsCount).map {
         DefaultKtorRequestsExecutor(
@@ -66,7 +66,7 @@ class MultipleClientKtorRequestsExecutor (
     private val clientAllocationMutex = Mutex()
     private val takerFlow = freeClients.mapNotNull {
         clientAllocationMutex.withLock {
-            freeClients.value.firstOrNull() ?.also {
+            freeClients.value.firstOrNull()?.also {
                 freeClients.value -= it
             } ?: return@mapNotNull null
         }
@@ -81,7 +81,7 @@ class MultipleClientKtorRequestsExecutor (
         jsonFormatter: Json,
         pipelineStepsHolder: TelegramBotPipelinesHandler,
         logger: KSLog,
-        diff: Unit
+        diff: Unit,
     ) : this(
         telegramAPIUrlsKeeper,
         callsFactories,
@@ -89,7 +89,7 @@ class MultipleClientKtorRequestsExecutor (
         requestsLimiter,
         jsonFormatter,
         pipelineStepsHolder,
-        client.engineConfig.threadsCount,
+        requestExecutorsCount = 4, // default threads count; configurable through dispatcher property
         logger,
         { platformClientCopy(client) }
     )

tgbotapi.core/src/jsMain/kotlin/dev/inmo/tgbotapi/utils/asInput.kt

@@ -1,8 +1,6 @@
 package dev.inmo.tgbotapi.utils
 
-import io.ktor.util.toByteArray
+import io.ktor.utils.io.*
-import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.core.*
-import io.ktor.utils.io.core.ByteReadPacket
-import io.ktor.utils.io.core.Input
 
 actual suspend fun ByteReadChannel.asInput(): Input = ByteReadPacket(toByteArray())

tgbotapi.utils/src/jvmMain/kotlin/dev/inmo/tgbotapi/extensions/utils/updates/retrieving/Webhook.kt

@@ -1,19 +1,21 @@
 package dev.inmo.tgbotapi.extensions.utils.updates.retrieving
 
-import dev.inmo.micro_utils.coroutines.*
+import dev.inmo.micro_utils.coroutines.ExceptionHandler
+import dev.inmo.micro_utils.coroutines.runCatchingSafely
 import dev.inmo.tgbotapi.bot.RequestsExecutor
 import dev.inmo.tgbotapi.extensions.utils.nonstrictJsonFormat
 import dev.inmo.tgbotapi.extensions.utils.updates.flowsUpdatesFilter
 import dev.inmo.tgbotapi.requests.webhook.SetWebhookRequest
 import dev.inmo.tgbotapi.types.update.abstracts.Update
 import dev.inmo.tgbotapi.types.update.abstracts.UpdateDeserializationStrategy
-import dev.inmo.tgbotapi.updateshandlers.*
+import dev.inmo.tgbotapi.updateshandlers.FlowsUpdatesFilter
+import dev.inmo.tgbotapi.updateshandlers.UpdateReceiver
+import dev.inmo.tgbotapi.updateshandlers.UpdatesFilter
 import dev.inmo.tgbotapi.updateshandlers.webhook.WebhookPrivateKeyConfig
-import io.ktor.http.HttpStatusCode
+import io.ktor.http.*
-import io.ktor.server.application.call
 import io.ktor.server.engine.*
-import io.ktor.server.request.receiveText
+import io.ktor.server.request.*
-import io.ktor.server.response.respond
+import io.ktor.server.response.*
 import io.ktor.server.routing.*
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.asCoroutineDispatcher
@@ -38,7 +40,7 @@ fun Route.includeWebhookHandlingInRoute(
     scope: CoroutineScope,
     exceptionsHandler: ExceptionHandler<Unit>? = null,
     mediaGroupsDebounceTimeMillis: Long = 1000L,
-    block: UpdateReceiver<Update>
+    block: UpdateReceiver<Update>,
 ) {
     val transformer = scope.updateHandlerWithMediaGroupsAdaptation(block, mediaGroupsDebounceTimeMillis)
     post {
@@ -55,7 +57,7 @@ fun Route.includeWebhookHandlingInRoute(
                 call.respond(HttpStatusCode.InternalServerError)
             }.getOrThrow()
         } catch (e: Throwable) {
-            exceptionsHandler ?.invoke(e)
+            exceptionsHandler?.invoke(e)
         }
     }
 }
@@ -69,7 +71,7 @@ fun Route.includeWebhookHandlingInRouteWithFlows(
     scope: CoroutineScope,
     exceptionsHandler: ExceptionHandler<Unit>? = null,
     mediaGroupsDebounceTimeMillis: Long = 1000L,
-    block: FlowsUpdatesFilter.() -> Unit
+    block: FlowsUpdatesFilter.() -> Unit,
 ) = includeWebhookHandlingInRoute(
     scope,
     exceptionsHandler,
@@ -92,50 +94,57 @@ fun Route.includeWebhookHandlingInRouteWithFlows(
  * @see UpdatesFilter
  * @see UpdatesFilter.asUpdateReceiver
  */
-fun startListenWebhooks(
+fun <TEngine : ApplicationEngine, TConfiguration : ApplicationEngine.Configuration> startListenWebhooks(
     listenPort: Int,
-    engineFactory: ApplicationEngineFactory<*, *>,
+    engineFactory: ApplicationEngineFactory<TEngine, TConfiguration>,
     exceptionsHandler: ExceptionHandler<Unit>,
     listenHost: String = "0.0.0.0",
     listenRoute: String? = null,
     privateKeyConfig: WebhookPrivateKeyConfig? = null,
     scope: CoroutineScope = CoroutineScope(Executors.newFixedThreadPool(4).asCoroutineDispatcher()),
     mediaGroupsDebounceTimeMillis: Long = 1000L,
-    additionalApplicationEngineEnvironmentConfigurator: ApplicationEngineEnvironmentBuilder.() -> Unit = {},
+    additionalApplicationEnvironmentConfigurator: ApplicationEnvironmentBuilder.() -> Unit = {},
-    block: UpdateReceiver<Update>
+    additionalEngineConfigurator: TConfiguration.() -> Unit = {},
-): ApplicationEngine {
+    block: UpdateReceiver<Update>,
-    val env = applicationEngineEnvironment {
+): EmbeddedServer<TEngine, TConfiguration> =
-
+    embeddedServer(
-        module {
+        factory = engineFactory,
-            routing {
+        environment = applicationEnvironment {
-                listenRoute ?.also {
+            additionalApplicationEnvironmentConfigurator()
-                    createRouteFromPath(it).includeWebhookHandlingInRoute(scope, exceptionsHandler, mediaGroupsDebounceTimeMillis, block)
+        },
-                } ?: includeWebhookHandlingInRoute(scope, exceptionsHandler, mediaGroupsDebounceTimeMillis, block)
+        configure = {
-            }
+            privateKeyConfig?.let {
-        }
+                sslConnector(
-
+                    privateKeyConfig.keyStore,
-        privateKeyConfig ?.let {
+                    privateKeyConfig.aliasName,
-            sslConnector(
+                    privateKeyConfig::keyStorePassword,
-                privateKeyConfig.keyStore,
+                    privateKeyConfig::aliasPassword
-                privateKeyConfig.aliasName,
+                ) {
-                privateKeyConfig::keyStorePassword,
+                    host = listenHost
-                privateKeyConfig::aliasPassword
+                    port = listenPort
-            ) {
+                }
+            } ?: connector {
                 host = listenHost
                 port = listenPort
             }
-        } ?: connector {
+
-            host = listenHost
+            additionalEngineConfigurator()
-            port = listenPort
+        },
+        module = {
+            routing {
+                listenRoute?.also {
+                    createRouteFromPath(it).includeWebhookHandlingInRoute(
+                        scope,
+                        exceptionsHandler,
+                        mediaGroupsDebounceTimeMillis,
+                        block
+                    )
+                } ?: includeWebhookHandlingInRoute(scope, exceptionsHandler, mediaGroupsDebounceTimeMillis, block)
+            }
         }
-
+    ).also {
-        additionalApplicationEngineEnvironmentConfigurator()
-    }
-
-    return embeddedServer(engineFactory, env).also {
         it.start(false)
     }
-}
 
 /**
  * Setting up ktor server, set webhook info via [SetWebhookRequest] request.
@@ -152,9 +161,9 @@ fun startListenWebhooks(
  * @see UpdatesFilter.asUpdateReceiver
  */
 @Suppress("unused")
-suspend fun RequestsExecutor.setWebhookInfoAndStartListenWebhooks(
+suspend fun <TEngine : ApplicationEngine, TConfiguration : ApplicationEngine.Configuration> RequestsExecutor.setWebhookInfoAndStartListenWebhooks(
     listenPort: Int,
-    engineFactory: ApplicationEngineFactory<*, *>,
+    engineFactory: ApplicationEngineFactory<TEngine, TConfiguration>,
     setWebhookRequest: SetWebhookRequest,
     exceptionsHandler: ExceptionHandler<Unit> = {},
     listenHost: String = "0.0.0.0",
@@ -162,11 +171,24 @@ suspend fun RequestsExecutor.setWebhookInfoAndStartListenWebhooks(
     privateKeyConfig: WebhookPrivateKeyConfig? = null,
     scope: CoroutineScope = CoroutineScope(Executors.newFixedThreadPool(4).asCoroutineDispatcher()),
     mediaGroupsDebounceTimeMillis: Long = 1000L,
-    additionalApplicationEngineEnvironmentConfigurator: ApplicationEngineEnvironmentBuilder.() -> Unit = {},
+    additionalApplicationEnvironmentConfigurator: ApplicationEnvironmentBuilder.() -> Unit = {},
-    block: UpdateReceiver<Update>
+    additionalEngineConfigurator: TConfiguration.() -> Unit = {},
-): ApplicationEngine = try {
+    block: UpdateReceiver<Update>,
+): EmbeddedServer<TEngine, TConfiguration> = try {
     execute(setWebhookRequest)
-    startListenWebhooks(listenPort, engineFactory, exceptionsHandler, listenHost, listenRoute, privateKeyConfig, scope, mediaGroupsDebounceTimeMillis, additionalApplicationEngineEnvironmentConfigurator, block)
+    startListenWebhooks(
+        listenPort,
+        engineFactory,
+        exceptionsHandler,
+        listenHost,
+        listenRoute,
+        privateKeyConfig,
+        scope,
+        mediaGroupsDebounceTimeMillis,
+        additionalApplicationEnvironmentConfigurator,
+        additionalEngineConfigurator,
+        block
+    )
 } catch (e: Exception) {
     throw e
 }