Refactor NewsService and SessionizeService to use CoroutineScope for job lifecycle management and adjust DI module accordingly. (#531)

Co-authored-by: Leonid Stashevsky <[email protected]>

Author: nomisRev

backend/src/main/kotlin/org/jetbrains/kotlinconf/backend/DiModule.kt

@@ -4,45 +4,46 @@ import io.ktor.client.*
 import io.ktor.client.plugins.contentnegotiation.*
 import io.ktor.serialization.kotlinx.json.*
 import io.ktor.server.application.*
-import io.ktor.utils.io.core.*
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.serialization.json.Json
 import org.jetbrains.kotlinconf.backend.di.configModule
 import org.jetbrains.kotlinconf.backend.di.repositoryModule
 import org.jetbrains.kotlinconf.backend.di.serviceModule
 import org.koin.dsl.module
+import org.koin.dsl.onClose
 import org.koin.ktor.ext.getKoin
 import org.koin.ktor.plugin.Koin
 
 fun Application.diModule() {
-    val client = HttpClient {
-        install(ContentNegotiation) {
-            json(Json {
-                ignoreUnknownKeys = true
-
-                encodeDefaults = true
-                isLenient = true
-                allowSpecialFloatingPointValues = true
-                allowStructuredMapKeys = true
-                prettyPrint = false
-                useArrayPolymorphism = false
-            })
-        }
-    }
-
     install(Koin) {
         modules(
             serviceModule,
             repositoryModule,
             configModule,
             module {
-                single { client }
+                single<CoroutineScope> { this@diModule }
+                single { createClient() } onClose { it?.close() }
                 single { environment.config }
             }
         )
     }
 
-    monitor.subscribe(ApplicationStopping) {
-        val closeableServices = it.getKoin().getAll<Closeable>()
-        closeableServices.forEach { it.close() }
+    monitor.subscribe(ApplicationStopped) {
+        it.getKoin().close()
+    }
+}
+
+private fun createClient() = HttpClient {
+    install(ContentNegotiation) {
+        json(Json {
+            ignoreUnknownKeys = true
+
+            encodeDefaults = true
+            isLenient = true
+            allowSpecialFloatingPointValues = true
+            allowStructuredMapKeys = true
+            prettyPrint = false
+            useArrayPolymorphism = false
+        })
     }
-}
+}

backend/src/main/kotlin/org/jetbrains/kotlinconf/backend/services/NewsService.kt

@@ -4,8 +4,9 @@ import io.ktor.client.*
 import io.ktor.client.call.*
 import io.ktor.client.request.*
 import io.ktor.utils.io.core.*
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
@@ -15,7 +16,11 @@ import org.jetbrains.kotlinconf.backend.model.GitHubItem
 import org.jetbrains.kotlinconf.backend.utils.ConferenceConfig
 import org.slf4j.LoggerFactory
 
-class NewsService(private val client: HttpClient, config: ConferenceConfig): Closeable {
+class NewsService(
+    private val client: HttpClient,
+    scope: CoroutineScope,
+    config: ConferenceConfig
+) {
     private val log = LoggerFactory.getLogger("NewsService")
     private val repo: String = config.newsRepo
     private val branch: String = config.newsBranch
@@ -24,13 +29,15 @@ class NewsService(private val client: HttpClient, config: ConferenceConfig): Clo
 
     private val news = MutableSharedFlow<List<NewsItem>>(replay = 1)
 
-    val syncJob = GlobalScope.launch(Dispatchers.IO) {
-        while (true) {
-            runCatching { updateNews() }.onFailure {
-                log.error("Failed to update news", it)
-            }
+    init {
+        scope.launch(Dispatchers.IO) {
+            while (true) {
+                runCatching { updateNews() }.onFailure {
+                    log.error("Failed to update news", it)
+                }
 
-            kotlinx.coroutines.delay(updateInterval * 1000)
+                delay(updateInterval * 1000)
+            }
         }
     }
 
@@ -127,9 +134,4 @@ class NewsService(private val client: HttpClient, config: ConferenceConfig): Clo
             content = bodyContent
         )
     }
-
-
-    override fun close() {
-        syncJob.cancel()
-    }
 }

backend/src/main/kotlin/org/jetbrains/kotlinconf/backend/services/SessionizeService.kt

@@ -5,9 +5,7 @@ package org.jetbrains.kotlinconf.backend.services
 import io.ktor.client.HttpClient
 import io.ktor.client.call.body
 import io.ktor.client.request.get
-import io.ktor.utils.io.core.Closeable
-import kotlinx.coroutines.DelicateCoroutinesApi
-import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.first
@@ -26,27 +24,29 @@ import kotlin.time.ExperimentalTime
 
 class SessionizeService(
     private val client: HttpClient,
+    scope: CoroutineScope,
     config: ConferenceConfig
-): Closeable {
+) {
     private val conference = MutableSharedFlow<Conference>(replay = 1)
 
     private val sessionizeUrl: String = config.sessionizeUrl
     private val sessionizeInterval: Long = config.sessionizeInterval
     private val log = LoggerFactory.getLogger("SessionizeService")
 
-    @OptIn(DelicateCoroutinesApi::class)
-    private val syncJob = GlobalScope.launch {
-        log.info("Synchronizing each $sessionizeInterval minutes with $sessionizeUrl")
-        while (true) {
-            log.trace("Synchronizing to Sessionize…")
-            runCatching {
-                synchronizeWithSessionize(sessionizeUrl)
-            }.onFailure { cause ->
-                log.error("Failed to synchronize to Sessionize: ${cause.message}", cause)
+    init {
+        scope.launch {
+            log.info("Synchronizing each $sessionizeInterval minutes with $sessionizeUrl")
+            while (true) {
+                log.trace("Synchronizing to Sessionize…")
+                runCatching {
+                    synchronizeWithSessionize(sessionizeUrl)
+                }.onFailure { cause ->
+                    log.error("Failed to synchronize to Sessionize: ${cause.message}", cause)
+                }
+
+                log.trace("Finished loading data from Sessionize.")
+                delay(TimeUnit.MINUTES.toMillis(sessionizeInterval))
             }
-
-            log.trace("Finished loading data from Sessionize.")
-            delay(TimeUnit.MINUTES.toMillis(sessionizeInterval))
         }
     }
 
@@ -134,8 +134,4 @@ class SessionizeService(
 
         return nonWorkshop + workshops
     }
-
-    override fun close() {
-        syncJob.cancel()
-    }
 }