Use proper interrupt support on IO

RADAR-base · Oct 11, 2023 · 6c39236 · 6c39236
1 parent 06de816
commit 6c39236
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 10 deletions.
diff --git a/radar-commons-kotlin/README.md b/radar-commons-kotlin/README.md
@@ -10,7 +10,7 @@ First of all, it contains a cache implementation to use with coroutines, with su
 import java.io.IOException
 
 val dirCache = CachedMap<String, FileInfo> {
-    withContext(Dispatchers.IO) {
+    runInterruptible(Dispatchers.IO) {
         client.fetchDirectoryMap()
     }
 }

diff --git a/radar-commons-kotlin/src/main/kotlin/org/radarbase/kotlin/coroutines/Extensions.kt b/radar-commons-kotlin/src/main/kotlin/org/radarbase/kotlin/coroutines/Extensions.kt
@@ -11,8 +11,8 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.consume
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.runInterruptible
 import kotlinx.coroutines.sync.Semaphore
-import kotlinx.coroutines.withContext
 import java.util.concurrent.Future
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.CoroutineContext
@@ -49,15 +49,13 @@ suspend fun <T> Future<T>.suspendGet(
         }
     }
     try {
-        withContext(Dispatchers.IO) {
+        runInterruptible(Dispatchers.IO) {
             if (duration != null) {
                 get(duration.inWholeMilliseconds, TimeUnit.MILLISECONDS)
             } else {
                 get()
             }
         }
-    } catch (ex: InterruptedException) {
-        throw CancellationException("Future was interrupted", ex)
     } finally {
         channel.send(Unit)
     }

diff --git a/radar-commons-testing/src/test/java/org/radarbase/mock/data/MockRecordValidatorTest.kt b/radar-commons-testing/src/test/java/org/radarbase/mock/data/MockRecordValidatorTest.kt
@@ -16,8 +16,8 @@
 package org.radarbase.mock.data
 
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runInterruptible
 import kotlinx.coroutines.test.runTest
-import kotlinx.coroutines.withContext
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.io.TempDir
@@ -56,7 +56,7 @@ class MockRecordValidatorTest {
     fun validateEnum() = runTest {
         val config = makeConfig()
         config.valueSchema = ApplicationServerStatus::class.java.name
-        withContext(Dispatchers.IO) {
+        runInterruptible(Dispatchers.IO) {
             config.getDataFile(root).bufferedWriter().use { writer ->
                 writer.append("key.projectId,key.userId,key.sourceId,value.time,value.serverStatus,value.ipAddress\n")
                 writer.append("test,a,b,1,UNKNOWN,\n")
@@ -199,7 +199,7 @@ class MockRecordValidatorTest {
 
     private suspend fun writeConfig(write: Writer.() -> Unit): MockDataConfig {
         val config = makeConfig()
-        withContext(Dispatchers.IO) {
+        runInterruptible(Dispatchers.IO) {
             config.getDataFile(root).bufferedWriter().use(write)
         }
         return config

diff --git a/radar-commons/src/main/java/org/radarbase/producer/rest/RestKafkaSender.kt b/radar-commons/src/main/java/org/radarbase/producer/rest/RestKafkaSender.kt
@@ -41,6 +41,7 @@ import io.ktor.util.reflect.TypeInfo
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.withContext
@@ -181,11 +182,11 @@ class RestKafkaSender(config: Config) : KafkaSender {
             return true
         }
         val lastState = try {
-            val response = withContext(Dispatchers.IO) {
+            val response = scope.async {
                 restClient.head {
                     url("")
                 }
-            }
+            }.await()
             if (response.status.isSuccess()) {
                 _connectionState.didConnect()
                 ConnectionState.State.CONNECTED