Oppdatert avhengigheter, lagt til async lib

Author: naviktthomas

Diff for: apps/bekreftelse-api/src/test/kotlin/no/nav/paw/bekreftelse/api/producer/KafkaTestDataProducer.kt

@@ -37,5 +37,5 @@ fun main() {
         gjelderTil = Instant.now().plus(Duration.ofDays(14)),
     )
 
-    kafkaProducer.send(ProducerRecord(topic, key, value))
+    kafkaProducer.send(ProducerRecord(topic, key, value)).get()
 }

Diff for: gradle/libs.versions.toml

@@ -8,14 +8,14 @@ bekreftelseSchemaVersion = "1.25.03.11.31-1"
 arrowVersion = "2.0.1"
 arrowJacksonIntegrationVersion = "0.15.1"
 comSksamuelHopliteVersion = "2.9.0"
-graphqlClientVersion = "8.3.0"
+graphqlClientVersion = "8.4.0"
 orgApacheKafkaVersion = "3.9.0"
 ioConfluentKafkaVersion = "7.9.0"
 orgApacheAvroVersion = "1.12.0"
 comFasterxmlJacksonVersion = "2.18.3"
 kotlinExposedVersion = "0.60.0"
 logstashVersion = "8.0"
-logbackVersion = "1.5.17"
+logbackVersion = "1.5.18"
 kotestVersion = "5.9.1"
 mockkVersion = "1.13.17"
 testContainersVersion = "1.20.6"

Diff for: lib/async/build.gradle.kts

@@ -0,0 +1,23 @@
+plugins {
+    kotlin("jvm")
+}
+
+val jvmMajorVersion: String by project
+
+dependencies {
+    compileOnly(libs.kotlinx.coroutines.core)
+    compileOnly(libs.logbackClassic)
+
+    testImplementation(libs.test.junit5.runner)
+    testImplementation(libs.test.kotest.assertionsCore)
+}
+
+java {
+    toolchain {
+        languageVersion.set(JavaLanguageVersion.of(jvmMajorVersion))
+    }
+}
+
+tasks.withType<Test>().configureEach {
+    useJUnitPlatform()
+}

Diff for: lib/async/src/main/kotlin/no/nav/paw/async/runner/AsyncRunner.kt

@@ -0,0 +1,6 @@
+package no.nav.paw.async.runner
+
+interface AsyncRunner<T, R> {
+    fun run(task: () -> T, onFailure: (Throwable) -> Unit, onSuccess: (T) -> Unit): R
+    fun abort(onAbort: () -> Unit)
+}

Diff for: lib/async/src/main/kotlin/no/nav/paw/async/runner/CoroutineAsyncRunner.kt

@@ -0,0 +1,41 @@
+package no.nav.paw.async.runner
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+import org.slf4j.LoggerFactory
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
+
+open class CoroutineAsyncRunner<T>(
+    private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.IO),
+    private val coroutineDispatcher: CoroutineDispatcher = Dispatchers.IO,
+    private val keepRunning: AtomicBoolean = AtomicBoolean(true)
+) : AsyncRunner<T, Job> {
+    private val logger = LoggerFactory.getLogger(this.javaClass)
+    private val jobRef: AtomicReference<Job> = AtomicReference(Job())
+
+    override fun run(task: () -> T, onFailure: (Throwable) -> Unit, onSuccess: (T) -> Unit): Job {
+        logger.info("Running coroutine async function")
+        jobRef.set(coroutineScope.launch(coroutineDispatcher) {
+            do {
+                try {
+                    val result = task()
+                    onSuccess(result)
+                } catch (throwable: Throwable) {
+                    onFailure(throwable)
+                }
+            } while (keepRunning.get())
+        })
+        return jobRef.get()
+    }
+
+    override fun abort(onAbort: () -> Unit) {
+        logger.info("Aborting coroutine async function")
+        keepRunning.set(false)
+        jobRef.get().cancel()
+        onAbort()
+    }
+}

Diff for: lib/async/src/main/kotlin/no/nav/paw/async/runner/ScheduledAsyncRunner.kt

@@ -0,0 +1,49 @@
+package no.nav.paw.async.runner
+
+import org.slf4j.LoggerFactory
+import java.time.Duration
+import java.util.*
+import java.util.concurrent.atomic.AtomicBoolean
+
+open class ScheduledAsyncRunner<T>(
+    private val interval: Duration,
+    private val delay: Duration = Duration.ZERO,
+    private val keepRunning: AtomicBoolean = AtomicBoolean(true)
+) : AsyncRunner<T, Unit> {
+    private val logger = LoggerFactory.getLogger(this.javaClass)
+    private val timer = Timer()
+
+    override fun run(task: () -> T, onFailure: (Throwable) -> Unit, onSuccess: (T) -> Unit) {
+        logger.info("Starting scheduled async function with interval {} and delay {}", interval, delay)
+        val timerTask = ScheduledTimerTask(task, onSuccess, onFailure, keepRunning)
+        timer.scheduleAtFixedRate(timerTask, delay.toMillis(), interval.toMillis())
+    }
+
+    override fun abort(onAbort: () -> Unit) {
+        logger.info("Aborting scheduled async function")
+        keepRunning.set(false)
+        timer.cancel()
+        onAbort()
+    }
+
+    class ScheduledTimerTask<T>(
+        private val task: () -> T,
+        private val onSuccess: (T) -> Unit,
+        private val onFailure: (Throwable) -> Unit,
+        private val keepRunning: AtomicBoolean = AtomicBoolean(true)
+    ) : TimerTask() {
+        private val logger = LoggerFactory.getLogger(this.javaClass)
+
+        override fun run() {
+            logger.info("Running scheduled async function")
+            do {
+                try {
+                    val result = task()
+                    onSuccess(result)
+                } catch (throwable: Throwable) {
+                    onFailure(throwable)
+                }
+            } while (keepRunning.get())
+        }
+    }
+}

Diff for: lib/async/src/main/kotlin/no/nav/paw/async/runner/ThreadPoolAsyncRunner.kt

@@ -0,0 +1,40 @@
+package no.nav.paw.async.runner
+
+import org.slf4j.LoggerFactory
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
+
+open class ThreadPoolAsyncRunner<T>(
+    private val executorService: ExecutorService = Executors.newSingleThreadExecutor(),
+    private val keepRunning: AtomicBoolean = AtomicBoolean(true),
+    private val meyInterruptIfRunning: AtomicBoolean = AtomicBoolean(true)
+) : AsyncRunner<T, Future<*>> {
+    private val logger = LoggerFactory.getLogger(this.javaClass)
+    private val futureRef: AtomicReference<Future<*>> = AtomicReference(CompletableFuture<Nothing>())
+
+    override fun run(task: () -> T, onFailure: (Throwable) -> Unit, onSuccess: (T) -> Unit): Future<*> {
+        logger.info("Running thread pool async function")
+        futureRef.set(executorService.submit {
+            do {
+                try {
+                    val result = task()
+                    onSuccess(result)
+                } catch (throwable: Throwable) {
+                    onFailure(throwable)
+                }
+            } while (keepRunning.get())
+        })
+        return futureRef.get()
+    }
+
+    override fun abort(onAbort: () -> Unit) {
+        logger.info("Aborting thread pool async function")
+        keepRunning.set(false)
+        futureRef.get().cancel(meyInterruptIfRunning.get())
+        onAbort()
+    }
+}

Diff for: lib/scheduling/build.gradle.kts

@@ -3,6 +3,7 @@ plugins {
 }
 
 dependencies {
+    api(project(":lib:async"))
     compileOnly(libs.ktor.server.core)
     compileOnly(libs.kotlinx.coroutines.core)
 

Diff for: lib/scheduling/src/main/kotlin/no/nav/paw/scheduling/plugin/ScheduledTaskPlugin.kt

@@ -8,24 +8,17 @@ import io.ktor.server.application.ApplicationStopping
 import io.ktor.server.application.createApplicationPlugin
 import io.ktor.server.application.hooks.MonitoringEvent
 import io.ktor.server.application.log
-import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.launch
-import org.slf4j.LoggerFactory
+import no.nav.paw.async.runner.ScheduledAsyncRunner
 import java.time.Duration
-import java.util.*
 
-private val logger = LoggerFactory.getLogger("no.nav.paw.logger.scheduling")
 private const val PLUGIN_NAME_SUFFIX = "ScheduledTaskPlugin"
 
 class ScheduledTaskPluginConfig {
     var task: (() -> Unit)? = null
+    var interval: Duration? = null
     var delay: Duration? = null
-    var period: Duration? = null
     var startEvent: EventDefinition<Application>? = null
     var stopEvent: EventDefinition<Application>? = null
-    var coroutineDispatcher: CoroutineDispatcher? = null
 }
 
 @Suppress("FunctionName")
@@ -34,30 +27,18 @@ fun ScheduledTaskPlugin(pluginInstance: Any): ApplicationPlugin<ScheduledTaskPlu
     return createApplicationPlugin(pluginName, ::ScheduledTaskPluginConfig) {
         application.log.info("Installerer {}", pluginName)
         val task = requireNotNull(pluginConfig.task) { "Task er null" }
+        val interval = requireNotNull(pluginConfig.interval) { "Interval er null" }
         val delay = requireNotNull(pluginConfig.delay) { "Delay er null" }
-        val period = requireNotNull(pluginConfig.period) { "Period er null" }
         val startEvent = pluginConfig.startEvent ?: ApplicationStarted
         val stopEvent = pluginConfig.stopEvent ?: ApplicationStopping
-        val coroutineDispatcher = pluginConfig.coroutineDispatcher ?: Dispatchers.IO
-        var job: Job? = null
-        val timer = Timer()
-        val timerTask = object : TimerTask() {
-            override fun run() {
-                logger.info("Running scheduled task {}", pluginInstance)
-                task()
-            }
-        }
+        val asyncRunner = ScheduledAsyncRunner<Unit>(interval, delay)
 
-        on(MonitoringEvent(startEvent)) { application ->
-            application.log.info("Scheduling task {} at delay: {}, period {}", pluginInstance, delay, period)
-            job = application.launch(coroutineDispatcher) {
-                timer.scheduleAtFixedRate(timerTask, delay.toMillis(), period.toMillis())
-            }
+        on(MonitoringEvent(startEvent)) {
+            asyncRunner.run(task, {}) {}
         }
 
-        on(MonitoringEvent(stopEvent)) { _ ->
-            application.log.info("Canceling scheduled task {}", pluginInstance)
-            job?.cancel()
+        on(MonitoringEvent(stopEvent)) {
+            asyncRunner.abort {}
         }
     }
 }

Diff for: lib/scheduling/src/main/kotlin/no/nav/paw/scheduling/plugin/ScheduledTaskPluginInstaller.kt

@@ -2,25 +2,24 @@ package no.nav.paw.scheduling.plugin
 
 import io.ktor.events.EventDefinition
 import io.ktor.server.application.Application
+import io.ktor.server.application.ApplicationStarted
+import io.ktor.server.application.ApplicationStopping
 import io.ktor.server.application.install
-import kotlinx.coroutines.CoroutineDispatcher
 import java.time.Duration
 
 fun Application.installScheduledTaskPlugin(
     pluginInstance: Any,
     task: (() -> Unit),
-    delay: Duration? = null,
-    period: Duration? = null,
-    startEvent: EventDefinition<Application>? = null,
-    stopEvent: EventDefinition<Application>? = null,
-    coroutineDispatcher: CoroutineDispatcher? = null
+    interval: Duration,
+    delay: Duration = Duration.ZERO,
+    startEvent: EventDefinition<Application> = ApplicationStarted,
+    stopEvent: EventDefinition<Application> = ApplicationStopping
 ) {
     install(ScheduledTaskPlugin(pluginInstance)) {
         this.task = task
         this.delay = delay
-        this.period = period
+        this.interval = interval
         this.startEvent = startEvent
         this.stopEvent = stopEvent
-        this.coroutineDispatcher = coroutineDispatcher
     }
 }

Diff for: settings.gradle.kts

@@ -1,11 +1,11 @@
 plugins {
     id("org.gradle.toolchains.foojay-resolver-convention") version "0.9.0"
-    kotlin("jvm") version "2.1.10" apply false
-    kotlin("plugin.serialization") version "2.1.10" apply false
-    id("com.google.cloud.tools.jib") version "3.4.4" apply false
+    kotlin("jvm") version "2.1.20" apply false
+    kotlin("plugin.serialization") version "2.1.20" apply false
+    id("com.google.cloud.tools.jib") version "3.4.5" apply false
     id("org.openapi.generator") version "7.12.0" apply false
     id("com.github.davidmc24.gradle.plugin.avro") version "1.9.1" apply false
-    id("com.expediagroup.graphql") version "8.3.0" apply false
+    id("com.expediagroup.graphql") version "8.4.0" apply false
 }
 
 rootProject.name = "paw-arbeidssoekerregisteret-monorepo-intern"
@@ -20,6 +20,7 @@ include(
     "lib:api-docs",
     "lib:security",
     "lib:database",
+    "lib:async",
     "lib:scheduling",
     "lib:http-client-utils",
     "lib:kafka",