changes timeouts to more reasonable values (#206)

* changes timeouts to more reasonable values * use streamClient's timeout values for timeout check
yorkie-team · Jun 14, 2024 · e2915b8 · e2915b8
1 parent 971eb30
commit e2915b8
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/yorkie/src/main/kotlin/dev/yorkie/core/Client.kt b/yorkie/src/main/kotlin/dev/yorkie/core/Client.kt
@@ -45,6 +45,7 @@ import kotlin.collections.Map.Entry
 import kotlin.coroutines.coroutineContext
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
@@ -118,6 +119,10 @@ public class Client @VisibleForTesting internal constructor(
     private val Document.mutex
         get() = mutexForDocuments.getOrPut(key) { Mutex() }
 
+    private val streamTimeout = with(streamClient) {
+        callTimeoutMillis.takeIf { it > 0 } ?: (connectTimeoutMillis + readTimeoutMillis)
+    }.takeIf { it > 0 }?.milliseconds ?: 5.minutes
+
     public constructor(
         host: String,
         options: Options = Options(),
@@ -312,7 +317,7 @@ public class Client @VisibleForTesting internal constructor(
             while (true) {
                 ensureActive()
                 latestStream.safeClose()
-                val stream = withTimeoutOrNull(1_000) {
+                val stream = withTimeoutOrNull(streamTimeout) {
                     service.watchDocument(
                         attachment.document.key.documentBasedRequestHeader,
                     ).also {
@@ -322,7 +327,7 @@ public class Client @VisibleForTesting internal constructor(
                 val streamJob = launch(start = CoroutineStart.UNDISPATCHED) {
                     val channel = stream.responseChannel()
                     while (!stream.isReceiveClosed() && !channel.isClosedForReceive) {
-                        withTimeoutOrNull(60_000) {
+                        withTimeoutOrNull(streamTimeout) {
                             val receiveResult = channel.receiveCatching()
                             receiveResult.onSuccess {
                                 attachment.document.publishEvent(StreamConnectionChanged.Connected)