add more channel resubscribing

Author: jan-tennert

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/Realtime.kt

@@ -127,12 +127,16 @@ interface Realtime : MainPlugin<Realtime.Config>, CustomSerializationPlugin {
      * @property disconnectOnNoSubscriptions Whether to disconnect from the websocket when there are no more subscriptions. Defaults to true
      * @property serializer A serializer used for serializing/deserializing objects e.g. in [PresenceAction.decodeJoinsAs] or [RealtimeChannel.broadcast]. Defaults to [KotlinXSerializer]
      * @property websocketFactory A custom websocket factory. If this is set, the [websocketConfig] will be ignored
+     * @property rejoinDelay The interval between channel rejoin attempts
+     * @property maxAttempts The maximum amount of connection attempts before giving up. Defaults to 5
      */
     data class Config(
         var websocketConfig: WebSockets.Config.() -> Unit = {},
         var secure: Boolean? = null,
         var heartbeatInterval: Duration = 15.seconds,
         var reconnectDelay: Duration = 7.seconds,
+        var rejoinDelay: Duration = 2.seconds,
+        var maxAttempts: Int = 5,
         var disconnectOnSessionLoss: Boolean = true,
         var connectOnSubscribe: Boolean = true,
         @property:SupabaseInternal var websocketFactory: RealtimeWebsocketFactory? = null,

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannel.kt

@@ -110,6 +110,9 @@ interface RealtimeChannel {
     @SupabaseInternal
     fun updateStatus(status: Status)
 
+    @SupabaseInternal
+    suspend fun scheduleRejoin()
+
     /**
      * Represents the status of a channel
      */

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt

@@ -11,6 +11,7 @@ import io.github.jan.supabase.realtime.event.RealtimeEvent
 import io.ktor.client.statement.bodyAsText
 import io.ktor.http.headers
 import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -25,6 +26,7 @@ import kotlinx.serialization.json.encodeToJsonElement
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.put
 import kotlinx.serialization.json.putJsonObject
+import kotlin.concurrent.atomics.AtomicInt
 import kotlin.concurrent.atomics.incrementAndFetch
 import kotlin.reflect.KClass
 import kotlin.reflect.KType
@@ -49,6 +51,8 @@ internal class RealtimeChannelImpl(
     private val subTopic = topic.replaceFirst(Regex("^${RealtimeTopic.PREFIX}:", RegexOption.IGNORE_CASE), "")
     private val httpClient = realtimeImpl.supabaseClient.httpClient
 
+    internal val joinAttempt = AtomicInt(0)
+
     private suspend fun accessToken() = realtimeImpl.config.accessToken(supabaseClient) ?: realtimeImpl.accessToken
 
     @OptIn(SupabaseInternal::class)
@@ -92,6 +96,12 @@ internal class RealtimeChannelImpl(
         event.handle(this, message)
     }
 
+    override suspend fun scheduleRejoin() {
+        Realtime.logger.d { "Rejoining channel $topic in" }
+        delay(realtime.config.rejoinDelay)
+        resubscribe()
+    }
+
     override suspend fun unsubscribe() {
         _status.value = RealtimeChannel.Status.UNSUBSCRIBING
         Realtime.logger.d { "Unsubscribing from channel $topic" }
@@ -232,6 +242,9 @@ internal class RealtimeChannelImpl(
     }
 
     override fun updateStatus(status: RealtimeChannel.Status) {
+        if(this.status.value == RealtimeChannel.Status.SUBSCRIBED) {
+            joinAttempt.store(0)
+        }
         _status.value = status
     }
 

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/event/RErrorEvent.kt

@@ -3,15 +3,25 @@ package io.github.jan.supabase.realtime.event
 import io.github.jan.supabase.logging.e
 import io.github.jan.supabase.realtime.Realtime
 import io.github.jan.supabase.realtime.RealtimeChannel
+import io.github.jan.supabase.realtime.RealtimeChannelImpl
 import io.github.jan.supabase.realtime.RealtimeMessage
+import kotlin.concurrent.atomics.incrementAndFetch
 
 /**
  * Event that handles an error event
  */
 data object RErrorEvent : RealtimeEvent {
 
     override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) {
-        Realtime.logger.e { "Received an error in channel ${message.topic}. That could be as a result of an invalid access token" }
+        channel as RealtimeChannelImpl
+        val currentAttempt = channel.joinAttempt.incrementAndFetch()
+        if(currentAttempt >= channel.realtime.config.maxAttempts) {
+            Realtime.logger.e { "Failed to rejoin channel ${message.topic} after $currentAttempt attempts. Giving up." }
+            channel.updateStatus(RealtimeChannel.Status.UNSUBSCRIBED)
+            return
+        }
+        Realtime.logger.e { "Received an error in channel ${message.topic}. That could be as a result of an invalid access token. Trying to rejoin the channel..." }
+        channel.scheduleRejoin()
     }
 
     override fun appliesTo(message: RealtimeMessage): Boolean {

gradle.properties

@@ -11,5 +11,5 @@ org.jetbrains.compose.experimental.jscanvas.enabled=true
 org.jetbrains.compose.experimental.wasm.enabled=true
 org.jetbrains.dokka.experimental.gradle.pluginMode=V2Enabled
 
-supabase-version = 3.2.4-dev
+supabase-version = 3.2.4
 base-group = io.github.jan-tennert.supabase