SSE Reconnect (#9)

expose maxRetries for reconnection attempts, showRetryEvents controller Ktor log level from config
IdanAizikNissim · Oct 25, 2024 · fdc6e81 · fdc6e81
1 parent 1ee18f4
commit fdc6e81
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 10 deletions.
diff --git a/shared/src/commonMain/kotlin/io/pocketbase/ClientConfig.kt b/shared/src/commonMain/kotlin/io/pocketbase/ClientConfig.kt
@@ -1,5 +1,6 @@
 package io.pocketbase
 
+import io.pocketbase.http.LogLevel
 import io.pocketbase.http.Protocol
 import kotlin.time.Duration.Companion.seconds
 
@@ -8,7 +9,9 @@ data class ClientConfig(
     val protocol: Protocol = Protocol.HTTP,
     val port: Int?,
     val lang: String = "en-US",
-    val reconnectionTime: Long = 30.seconds.inWholeMilliseconds,
+    val reconnectionTime: Long = 3.seconds.inWholeMilliseconds,
+    val maxReconnectionRetries: Long = Long.MAX_VALUE,
+    val logLevel: LogLevel = LogLevel.NONE,
 ) {
     companion object {
         fun from(url: String): ClientConfig {

diff --git a/shared/src/commonMain/kotlin/io/pocketbase/http/HttpClient.kt b/shared/src/commonMain/kotlin/io/pocketbase/http/HttpClient.kt
@@ -31,6 +31,7 @@ internal class HttpClient(
             protocol = config.protocol,
             port = config.port,
             lang = config.lang,
+            logLevel = config.logLevel,
             authStore = authStore,
         )
 

diff --git a/shared/src/commonMain/kotlin/io/pocketbase/http/HttpClientBuilder.kt b/shared/src/commonMain/kotlin/io/pocketbase/http/HttpClientBuilder.kt
@@ -3,7 +3,6 @@ package io.pocketbase.http
 import io.ktor.client.HttpClient
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.client.plugins.defaultRequest
-import io.ktor.client.plugins.logging.LogLevel
 import io.ktor.client.plugins.logging.Logging
 import io.ktor.client.plugins.sse.SSE
 import io.ktor.client.request.header
@@ -18,6 +17,7 @@ internal class HttpClientBuilder(
         protocol: Protocol,
         port: Int?,
         lang: String,
+        logLevel: LogLevel,
         authStore: AuthStore,
     ): HttpClient =
         factory.create { config ->
@@ -29,7 +29,7 @@ internal class HttpClientBuilder(
                 }
 
                 install(Logging) {
-                    level = LogLevel.ALL
+                    level = logLevel.toKtorLogLevel()
                 }
 
                 defaultRequest {
@@ -47,3 +47,12 @@ internal class HttpClientBuilder(
             }
         }
 }
+
+internal fun LogLevel.toKtorLogLevel(): io.ktor.client.plugins.logging.LogLevel =
+    when (this) {
+        LogLevel.ALL -> io.ktor.client.plugins.logging.LogLevel.ALL
+        LogLevel.HEADERS -> io.ktor.client.plugins.logging.LogLevel.HEADERS
+        LogLevel.BODY -> io.ktor.client.plugins.logging.LogLevel.BODY
+        LogLevel.INFO -> io.ktor.client.plugins.logging.LogLevel.INFO
+        LogLevel.NONE -> io.ktor.client.plugins.logging.LogLevel.NONE
+    }
diff --git a/shared/src/commonMain/kotlin/io/pocketbase/http/LogLevel.kt b/shared/src/commonMain/kotlin/io/pocketbase/http/LogLevel.kt
@@ -0,0 +1,9 @@
+package io.pocketbase.http
+
+enum class LogLevel {
+    ALL,
+    HEADERS,
+    BODY,
+    INFO,
+    NONE,
+}
diff --git a/shared/src/commonMain/kotlin/io/pocketbase/http/SSEClient.kt b/shared/src/commonMain/kotlin/io/pocketbase/http/SSEClient.kt
@@ -5,6 +5,7 @@ import io.ktor.client.plugins.sse.sse
 import io.ktor.sse.ServerSentEvent
 import io.pocketbase.ClientConfig
 import io.pocketbase.auth.AuthStore
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.flow
 import kotlin.time.Duration.Companion.milliseconds
 
@@ -13,6 +14,8 @@ internal class SSEClient(
     private val authStore: AuthStore,
 ) {
     private var client: HttpClient? = null
+    private var retryAttempts = 0
+
     var clientId: String? = null
         private set
 
@@ -29,22 +32,28 @@ internal class SSEClient(
                     protocol = config.protocol,
                     port = config.port,
                     lang = config.lang,
+                    logLevel = config.logLevel,
                     authStore = authStore,
                 )
 
             client?.sse(
                 path = url,
                 reconnectionTime = config.reconnectionTime.milliseconds,
+                showRetryEvents = true,
             ) {
                 incoming.collect { msg ->
-                    when (msg.event) {
-                        "PB_CONNECT" -> {
-                            clientId = msg.id
-                            Incoming.PBConnect(clientId)
+                    if (msg.retry != null) {
+                        reconnect(url, msg.retry ?: 0)
+                    } else {
+                        when (msg.event) {
+                            "PB_CONNECT" -> {
+                                clientId = msg.id
+                                Incoming.PBConnect(clientId)
+                            }
+                            else -> Incoming.Message(msg)
+                        }.let {
+                            emit(it)
                         }
-                        else -> Incoming.Message(msg)
-                    }.let {
-                        emit(it)
                     }
                 }
             }
@@ -55,6 +64,21 @@ internal class SSEClient(
         client = null
         clientId = null
     }
+
+    private suspend fun reconnect(
+        url: String,
+        retry: Long,
+    ) {
+        if (retryAttempts > config.maxReconnectionRetries) {
+            disconnect()
+            return
+        }
+
+        delay(retry)
+        retryAttempts++
+
+        connect(url)
+    }
 }
 
 internal sealed interface Incoming {