feat: migrate from webflux to ktor

Replaced the entire spring webflux/reactor pipeline with a ktor based http clients.

Spring Boot remains only for dependency injection and configuration.

Author: jboix

build.gradle.kts

@@ -26,12 +26,15 @@ repositories {
 
 dependencies {
   // Dependencies
-  implementation("org.springframework.boot:spring-boot-starter-webflux")
+  implementation("org.springframework.boot:spring-boot-starter-json")
   implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
   implementation("org.jetbrains.kotlin:kotlin-reflect")
   implementation("nl.basjes.parse.useragent:yauaa:7.32.0")
   implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
-  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
+  implementation("io.ktor:ktor-client-core:3.3.2")
+  implementation("io.ktor:ktor-client-cio:3.3.2")
+  implementation("io.ktor:ktor-client-content-negotiation:3.3.2")
+  implementation("io.ktor:ktor-serialization-jackson:3.3.2")
 
   // Test Dependencies
   testImplementation("io.kotest:kotest-runner-junit5:5.9.1")

src/main/kotlin/ch/srgssr/pillarbox/monitoring/DataTransferApplicationRunner.kt

@@ -2,13 +2,13 @@ package ch.srgssr.pillarbox.monitoring
 
 import ch.srgssr.pillarbox.monitoring.event.EventDispatcherClient
 import ch.srgssr.pillarbox.monitoring.event.setup.OpenSearchSetupService
+import ch.srgssr.pillarbox.monitoring.exception.HttpClientException
 import ch.srgssr.pillarbox.monitoring.log.logger
 import kotlinx.coroutines.runBlocking
 import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
 import org.springframework.context.annotation.Profile
 import org.springframework.stereotype.Component
-import org.springframework.web.reactive.function.client.WebClientResponseException
 
 /**
  * DataTransferApplicationRunner is responsible for initializing the OpenSearch setup and
@@ -48,13 +48,8 @@ class DataTransferApplicationRunner(
         openSearchSetupService.start()
         logger.info("All setup tasks are completed, starting SSE client...")
         eventDispatcherClient.start().join()
-      } catch (e: WebClientResponseException) {
-        logger.error(
-          "OpenSearch connection failed " +
-            "| [Status Code: ${e.statusCode.value()}] " +
-            "| [Body: ${e.responseBodyAsString}]",
-          e,
-        )
+      } catch (e: HttpClientException) {
+        logger.error("OpenSearch connection failed", e)
       } catch (e: Exception) {
         logger.error("OpenSearch setup failed due to an unexpected error", e)
       } finally {

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/EventDispatcherClient.kt

@@ -5,6 +5,7 @@ import ch.srgssr.pillarbox.monitoring.benchmark.timed
 import ch.srgssr.pillarbox.monitoring.cache.LRUCache
 import ch.srgssr.pillarbox.monitoring.event.model.EventRequest
 import ch.srgssr.pillarbox.monitoring.event.repository.EventRepository
+import ch.srgssr.pillarbox.monitoring.exception.HttpClientException
 import ch.srgssr.pillarbox.monitoring.flow.chunked
 import ch.srgssr.pillarbox.monitoring.log.info
 import ch.srgssr.pillarbox.monitoring.log.logger
@@ -18,7 +19,6 @@ import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import org.springframework.stereotype.Service
-import org.springframework.web.reactive.function.client.WebClientResponseException
 
 /**
  * Service responsible for consuming events from a remote event dispatcher service via Server-Sent Events (SSE),
@@ -98,13 +98,8 @@ class EventDispatcherClient(
       timed("EventRepository.saveEvents") {
         eventRepository.saveAll(events)
       }
-    } catch (e: WebClientResponseException) {
-      logger.error(
-        "A connection error occurred while saving the current batch " +
-          "| [Status Code: ${e.statusCode.value()}] " +
-          "| [Body: ${e.responseBodyAsString}]",
-        e,
-      )
+    } catch (e: HttpClientException) {
+      logger.error("An connection error occurred while saving the current batch", e)
     } catch (e: Exception) {
       logger.error("An unexpected error occurred while saving the current batch", e)
     }

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/EventFlowProvider.kt

@@ -1,12 +1,21 @@
 package ch.srgssr.pillarbox.monitoring.event
 
 import ch.srgssr.pillarbox.monitoring.event.model.EventRequest
+import ch.srgssr.pillarbox.monitoring.log.error
 import ch.srgssr.pillarbox.monitoring.log.logger
+import ch.srgssr.pillarbox.monitoring.log.warn
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.ktor.client.HttpClient
+import io.ktor.client.engine.cio.CIO
+import io.ktor.client.plugins.defaultRequest
+import io.ktor.client.plugins.sse.sse
+import io.ktor.http.ContentType
+import io.ktor.http.contentType
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.retryWhen
 import org.springframework.stereotype.Component
-import org.springframework.web.reactive.function.client.WebClient
-import org.springframework.web.reactive.function.client.bodyToFlow
 
 /**
  * Provides a reactive [Flow] of [EventRequest]s by connecting to a remote Server-Sent Events (SSE) endpoint.
@@ -17,12 +26,12 @@ import org.springframework.web.reactive.function.client.bodyToFlow
  * - Applying retry logic in case of transient failures, with support for logging retry attempts.
  *
  * @property properties The SSE client configuration including URI and retry strategy.
- * @property webClientBuilder A builder for constructing the [WebClient] used to make SSE requests.
+ * @property objectMapper
  */
 @Component
 class EventFlowProvider(
   private val properties: EventDispatcherClientConfiguration,
-  webClientBuilder: WebClient.Builder,
+  private val objectMapper: ObjectMapper,
 ) {
   private companion object {
     /**
@@ -31,26 +40,45 @@ class EventFlowProvider(
     private val logger = logger()
   }
 
-  private val webClient = webClientBuilder.baseUrl(properties.uri.toString()).build()
+  private val httpClient =
+    HttpClient(CIO) {
+      defaultRequest {
+        url(properties.uri.toString())
+        contentType(ContentType.Application.Json)
+      }
+    }
 
   /**
    * Creates and returns a [Flow] of [EventRequest]s from the SSE endpoint.
    *
    * @return A [Flow] that emits [EventRequest]s received from the remote SSE endpoint.
    */
+  @Suppress("TooGenericExceptionCaught")
   fun start(): Flow<EventRequest> =
-    webClient
-      .get()
-      .retrieve()
-      .bodyToFlow<EventRequest>()
-      .retryWhen(
-        properties.sseRetry.toRetryWhen(
-          onRetry = { cause, attempt, delayMillis ->
-            logger.warn(
-              "Retrying after failure: ${cause.message}. " +
-                "Attempt ${attempt + 1}. Waiting for ${delayMillis}ms",
-            )
-          },
-        ),
-      )
+    callbackFlow {
+      try {
+        httpClient.sse("/") {
+          incoming.collect {
+            val event = objectMapper.readValue(it.data, EventRequest::class.java)
+            trySend(event).isSuccess
+          }
+        }
+      } catch (e: Exception) {
+        logger.error(e) { "SSE connection failed: ${e.message}" }
+        close(e)
+      }
+
+      awaitClose {
+        logger.info("SSE flow closed")
+        httpClient.close()
+      }
+    }.retryWhen(
+      properties.sseRetry.toRetryWhen(
+        onRetry = { cause, attempt, delayMillis ->
+          logger.warn(cause) {
+            "Retrying after failure: ${cause.message}. Attempt ${attempt + 1}. Waiting for ${delayMillis}ms"
+          }
+        },
+      ),
+    )
 }

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/config/RetryProperties.kt

@@ -2,8 +2,6 @@ package ch.srgssr.pillarbox.monitoring.event.config
 
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.FlowCollector
-import reactor.util.retry.Retry
-import reactor.util.retry.RetryBackoffSpec
 import java.time.Duration
 
 /**
@@ -18,18 +16,6 @@ data class RetryProperties(
   val initialInterval: Duration = Duration.ofSeconds(5),
   val maxInterval: Duration = Duration.ofMinutes(1),
 ) {
-  /**
-   * Creates a [RetryBackoffSpec] based on the retry properties.
-   * This specification defines the backoff strategy for retries,
-   * including the number of attempts and interval timings.
-   *
-   * @return A configured [RetryBackoffSpec] for use with Reactor's retry mechanism.
-   */
-  fun create(): RetryBackoffSpec =
-    Retry
-      .backoff(maxAttempts, initialInterval)
-      .maxBackoff(maxInterval)
-
   /**
    * Creates a `retryWhen` callback function for Kotlin Flow.
    *

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/repository/EventRepository.kt

@@ -1,27 +1,30 @@
 package ch.srgssr.pillarbox.monitoring.event.repository
 
 import ch.srgssr.pillarbox.monitoring.event.model.EventRequest
+import ch.srgssr.pillarbox.monitoring.io.onSuccess
+import ch.srgssr.pillarbox.monitoring.io.throwOnNotSuccess
 import ch.srgssr.pillarbox.monitoring.log.debug
 import ch.srgssr.pillarbox.monitoring.log.error
 import ch.srgssr.pillarbox.monitoring.log.logger
 import com.fasterxml.jackson.databind.ObjectMapper
-import kotlinx.coroutines.reactor.awaitSingleOrNull
+import io.ktor.client.HttpClient
+import io.ktor.client.call.body
+import io.ktor.client.request.post
+import io.ktor.client.request.setBody
+import io.ktor.client.request.url
 import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.http.HttpHeaders
-import org.springframework.http.MediaType
 import org.springframework.stereotype.Component
-import org.springframework.web.reactive.function.client.WebClient
 
 /**
  * Repository responsible for sending event data to OpenSearch using the bulk API.
  *
- * @property webClient The [WebClient] configured to connect to the OpenSearch instance.
+ * @property httpClient The [HttpClient] configured to connect to the OpenSearch instance.
  * @property objectMapper Jackson's [ObjectMapper] used to serialize event objects.
  */
 @Component
 class EventRepository(
-  @param:Qualifier("openSearchWebClient")
-  private val webClient: WebClient,
+  @param:Qualifier("openSearchHttpClient")
+  private val httpClient: HttpClient,
   private val objectMapper: ObjectMapper,
 ) {
   private companion object {
@@ -43,15 +46,12 @@ class EventRepository(
    * @param events A list of events to be indexed.
    */
   suspend fun saveAll(events: List<EventRequest>) {
-    webClient
-      .post()
-      .uri("/_bulk")
-      .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
-      .bodyValue(toNDJson(events))
-      .retrieve()
-      .bodyToMono(String::class.java)
-      .awaitSingleOrNull()
-      ?.let { responseStr ->
+    httpClient
+      .post {
+        url("/_bulk")
+        setBody(toNDJson(events))
+      }.onSuccess {
+        val responseStr = body<String>()
         logger.debug { "Bulk response: $responseStr " }
 
         val response = objectMapper.readTree(responseStr)
@@ -67,7 +67,7 @@ class EventRepository(
               }
             }
         }
-      }
+      }.throwOnNotSuccess { "Connection error" }
   }
 
   /**

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/repository/OpenSearchConfigurationProperties.kt

@@ -3,7 +3,6 @@ package ch.srgssr.pillarbox.monitoring.event.repository
 import ch.srgssr.pillarbox.monitoring.event.config.RetryProperties
 import org.springframework.boot.context.properties.ConfigurationProperties
 import org.springframework.boot.context.properties.NestedConfigurationProperty
-import org.springframework.util.unit.DataSize
 import java.net.URI
 
 /**
@@ -14,14 +13,11 @@ import java.net.URI
  * @property uri The URI of the OpenSearch server. Defaults to `http://localhost:9200`.
  * @property retry Nested configuration properties for retry settings related to OpenSearch operations.
  * @property timeout The default timeout for each connection in milliseconds. 10s by default.
- * @property maxInMemorySize The maximum size in bytes allowed for buffering response bodies in memory,
- *                           64 MB by default.
  */
 @ConfigurationProperties(prefix = "pillarbox.monitoring.opensearch")
 data class OpenSearchConfigurationProperties(
   val uri: URI = URI("http://localhost:9200"),
   @NestedConfigurationProperty
   val retry: RetryProperties = RetryProperties(),
   val timeout: Int = 10_000,
-  val maxInMemorySize: DataSize = DataSize.ofMegabytes(64),
 )

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/AliasSetupTask.kt

@@ -1,34 +1,37 @@
 package ch.srgssr.pillarbox.monitoring.event.setup
 
+import ch.srgssr.pillarbox.monitoring.io.is4xxClientError
 import ch.srgssr.pillarbox.monitoring.io.loadResourceContent
-import ch.srgssr.pillarbox.monitoring.log.error
+import ch.srgssr.pillarbox.monitoring.io.onStatus
+import ch.srgssr.pillarbox.monitoring.io.onSuccess
+import ch.srgssr.pillarbox.monitoring.io.throwOnNotSuccess
 import ch.srgssr.pillarbox.monitoring.log.info
 import ch.srgssr.pillarbox.monitoring.log.logger
-import kotlinx.coroutines.reactor.awaitSingleOrNull
+import io.ktor.client.HttpClient
+import io.ktor.client.request.get
+import io.ktor.client.request.post
+import io.ktor.client.request.setBody
+import io.ktor.client.request.url
+import io.ktor.http.HttpStatusCode
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.core.annotation.Order
 import org.springframework.core.io.support.ResourcePatternResolver
-import org.springframework.http.HttpHeaders
-import org.springframework.http.HttpStatusCode
-import org.springframework.http.MediaType
 import org.springframework.stereotype.Component
-import org.springframework.web.reactive.function.client.WebClient
-import reactor.core.publisher.Mono
 
 /**
  * Task responsible for setting up a filtered alias in OpenSearch.
  *
  * This task checks if the specified alias exists in OpenSearch. If it does not, it loads the alias
  * configuration from `resources/opensearch/user_events-alias.json` and creates it.
  *
- * @property webClient WebClient instance used to interact with the OpenSearch API.
+ * @property httpClient [HttpClient] instance used to interact with the OpenSearch API.
  * @property resourceLoader Resource loader used to access the alias configuration JSON file.
  */
 @Component
 @Order(4)
 class AliasSetupTask(
-  @param:Qualifier("openSearchWebClient")
-  private val webClient: WebClient,
+  @param:Qualifier("openSearchHttpClient")
+  private val httpClient: HttpClient,
   private val resourceLoader: ResourcePatternResolver,
 ) : OpenSearchSetupTask {
   private companion object {
@@ -52,33 +55,23 @@ class AliasSetupTask(
     for (resource in resources) {
       val filename = resource.filename ?: continue
       val aliasName = filename.removeSuffix("-alias.json")
-      checkAndCreateAlias(aliasName).awaitSingleOrNull()
+      checkAndCreateAlias(aliasName)
     }
   }
 
-  private fun checkAndCreateAlias(aliasName: String): Mono<*> =
-    webClient
-      .get()
-      .uri("/_alias/$aliasName")
-      .retrieve()
+  private suspend fun checkAndCreateAlias(aliasName: String) =
+    httpClient
+      .get("/_alias/$aliasName")
       .onStatus(HttpStatusCode::is4xxClientError) {
-        logger.info { "Alias '$aliasName' does not exist, creating alias..." }
-        createAlias(aliasName).then(Mono.empty())
-      }.onStatus(HttpStatusCode::is2xxSuccessful) {
         logger.info { "Alias '$aliasName' already exists, skipping creation." }
-        Mono.empty()
-      }.toBodilessEntity()
+        createAlias(aliasName)
+      }.onSuccess { logger.info { "Alias '$aliasName' does not exist, creating alias..." } }
 
-  private fun createAlias(aliasName: String): Mono<*> {
-    val indexTemplateJson = resourceLoader.loadResourceContent("classpath:opensearch/$aliasName-alias.json")
-    return webClient
-      .post()
-      .uri("/_aliases")
-      .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
-      .bodyValue(indexTemplateJson)
-      .retrieve()
-      .toBodilessEntity()
-      .doOnSuccess { logger.info { "Alias '$aliasName' created successfully" } }
-      .doOnError { e -> logger.error { "Failed to create alias '$aliasName': ${e.message}" } }
-  }
+  private suspend fun createAlias(aliasName: String) =
+    httpClient
+      .post {
+        url("/_aliases")
+        setBody(resourceLoader.loadResourceContent("classpath:opensearch/$aliasName-alias.json"))
+      }.onSuccess { logger.info { "Alias '$aliasName' created successfully" } }
+      .throwOnNotSuccess { "Failed to create alias '$aliasName'" }
 }