Merge pull request #996 from supabase-community/presence-enabled

Add presence enabled flag

Author: jan-tennert

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

@@ -26,6 +26,8 @@ sealed interface CallbackManager {
 
     fun setServerChanges(changes: List<PostgresJoinConfig>)
 
+    fun getCallbacks(): List<RealtimeCallback<*>>
+
 }
 
 internal class CallbackManagerImpl(
@@ -37,6 +39,10 @@ internal class CallbackManagerImpl(
     val serverChanges: List<PostgresJoinConfig> get() = _serverChanges
     private val callbacks = AtomicMutableList<RealtimeCallback<*>>()
 
+    override fun getCallbacks(): List<RealtimeCallback<*>> {
+        return callbacks.toList()
+    }
+
     override fun addBroadcastCallback(event: String, callback: (JsonObject) -> Unit): Long {
         val id = nextId++
         callbacks += RealtimeCallback.BroadcastCallback(callback, event, id)

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

@@ -74,6 +74,12 @@ interface Realtime : MainPlugin<Realtime.Config>, CustomSerializationPlugin {
      */
     suspend fun removeChannel(channel: RealtimeChannel)
 
+    /**
+     * Adds a channel to the [subscriptions] without subscribing to it.
+     */
+    @SupabaseInternal
+    fun addChannel(channel: RealtimeChannel)
+
     /**
      * Unsubscribes and removes all channels from the [subscriptions]
      */

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

@@ -7,10 +7,12 @@ import io.github.jan.supabase.realtime.annotations.ChannelDsl
  * Used to build a realtime channel
  */
 @ChannelDsl
-class RealtimeChannelBuilder @PublishedApi internal constructor(private val topic: String) {
+class RealtimeChannelBuilder @PublishedApi internal constructor(
+    private val topic: String,
+) {
 
     private var broadcastJoinConfig = BroadcastJoinConfig(acknowledgeBroadcasts = false, receiveOwnBroadcasts = false)
-    private var presenceJoinConfig = PresenceJoinConfig("")
+    private var presenceJoinConfig = PresenceJoinConfig("", false)
 
     /**
      * Whether this channel should be private.
@@ -29,7 +31,7 @@ class RealtimeChannelBuilder @PublishedApi internal constructor(private val topi
      * @param block The presence join config
      */
     fun presence(block: PresenceJoinConfig.() -> Unit) {
-        presenceJoinConfig = PresenceJoinConfig("").apply(block)
+        presenceJoinConfig = PresenceJoinConfig("", false).apply(block)
     }
 
     @SupabaseInternal

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

@@ -17,6 +17,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
@@ -55,10 +56,15 @@ internal class RealtimeChannelImpl(
             if(!realtimeImpl.config.connectOnSubscribe) error("You can't subscribe to a channel while the realtime client is not connected. Did you forget to call `realtime.connect()`?")
             realtimeImpl.connect()
         }
+        if(!realtimeImpl.subscriptions.containsKey(topic)) {
+            realtime.addChannel(this)
+        }
         _status.value = RealtimeChannel.Status.SUBSCRIBING
         Realtime.logger.d { "Subscribing to channel $topic" }
         val currentJwt = accessToken()
         val postgrestChanges = clientChanges.toList()
+        val hasPresenceCallback = callbackManager.getCallbacks().filterIsInstance<RealtimeCallback.PresenceCallback>().isNotEmpty()
+        presenceJoinConfig.enabled = hasPresenceCallback
         val joinConfig = RealtimeJoinPayload(RealtimeJoinConfig(broadcastJoinConfig, presenceJoinConfig, postgrestChanges, isPrivate))
         val joinConfigObject = buildJsonObject {
             putJsonObject(Json.encodeToJsonElement(joinConfig).jsonObject)
@@ -214,12 +220,25 @@ internal class RealtimeChannelImpl(
             trySend(action)
         }
         val id = callbackManager.addPresenceCallback(callback)
+        if(status.value == RealtimeChannel.Status.SUBSCRIBED && !presenceJoinConfig.enabled) {
+            // If the channel is already subscribed, we need to resubscribe to enable presence
+            Realtime.logger.d { "Resubscribing to channel $topic to enable presence..." }
+            launch {
+                resubscribe()
+            }
+        }
         awaitClose { callbackManager.removeCallbackById(id) }
     }
 
     override fun updateStatus(status: RealtimeChannel.Status) {
         _status.value = status
     }
 
+    private suspend fun resubscribe() {
+        unsubscribe()
+        _status.first { it == RealtimeChannel.Status.UNSUBSCRIBED }
+        subscribe()
+    }
+
 }
 

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

@@ -93,6 +93,10 @@ import kotlin.time.Clock
         }
     }
 
+    override fun addChannel(channel: RealtimeChannel) {
+        _subscriptions[channel.topic] = channel
+    }
+
     override fun init() {
         scope.launch {
             supabaseClient.pluginManager.getPluginOrNull(Auth)?.sessionStatus?.collect {

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

@@ -30,9 +30,10 @@ data class BroadcastJoinConfig(@SerialName("ack") var acknowledgeBroadcasts: Boo
 
 /**
  * @param key Used to track presence payloads. Can be e.g. a user id
+ * @param enabled Whether presence is enabled for this channel
  */
 @Serializable
-data class PresenceJoinConfig(var key: String)
+data class PresenceJoinConfig(var key: String, internal var enabled: Boolean)
 
 @SupabaseInternal
 @Serializable

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

@@ -13,6 +13,7 @@ data object RCloseEvent : RealtimeEvent {
     override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) {
         channel.realtime.removeChannel(channel)
         Realtime.logger.d { "Unsubscribed from channel ${message.topic}" }
+        channel.updateStatus(RealtimeChannel.Status.UNSUBSCRIBED)
     }
 
     override fun appliesTo(message: RealtimeMessage): Boolean {

Realtime/src/commonTest/kotlin/CallbackManagerTest.kt

@@ -4,6 +4,7 @@ import io.github.jan.supabase.realtime.HasRecord
 import io.github.jan.supabase.realtime.PostgresAction
 import io.github.jan.supabase.realtime.PostgresJoinConfig
 import io.github.jan.supabase.realtime.Presence
+import io.github.jan.supabase.realtime.RealtimeCallback
 import io.github.jan.supabase.serializer.KotlinXSerializer
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
@@ -36,6 +37,18 @@ class CallbackManagerTest {
         assertFalse { called }
     }
 
+    @Test
+    fun testGetCallbacks() {
+        val cm = CallbackManagerImpl()
+        val expectedEvent = "event"
+        cm.addBroadcastCallback(expectedEvent) {
+            //...
+        }
+        val callbacks = cm.getCallbacks()
+        assertTrue { callbacks.isNotEmpty() }
+        assertTrue { callbacks.any { it is RealtimeCallback.BroadcastCallback && it.event == expectedEvent } }
+    }
+
     @Test
     fun testPresenceCallbacks() {
         val cm = CallbackManagerImpl()

Realtime/src/commonTest/kotlin/RealtimeChannelTest.kt

@@ -25,6 +25,7 @@ import io.ktor.client.engine.mock.respond
 import io.ktor.util.encodeBase64
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import kotlinx.serialization.json.Json
@@ -304,7 +305,7 @@ class RealtimeChannelTest {
         runTest {
             createTestClient(
                 wsHandler = { i, o ->
-                    handleSubscribe(i, o, channelId)
+                    handleSubscribe(i, o, channelId, true)
                     for(i in 0..amount) {
                         o.sendPresence(
                             channelId,
@@ -383,6 +384,38 @@ class RealtimeChannelTest {
         }
     }
 
+    @Test
+    fun testResubscribeOnPresenceChange() {
+        val channelId = "channelId"
+        runTest {
+            createTestClient(
+                wsHandler = { i, o ->
+                    handleSubscribe(i, o, channelId, false)
+                    handleUnsubscribe(i, o, channelId)
+                    handleSubscribe(i, o, channelId, true)
+                },
+                realtimeConfig = {
+                    disconnectOnNoSubscriptions = false
+                },
+                supabaseHandler = {
+                    val channel = it.channel(channelId)
+                    channel.status.test {
+                        assertEquals(RealtimeChannel.Status.UNSUBSCRIBED, awaitItem())
+                        channel.subscribe(false)
+                        assertEquals(RealtimeChannel.Status.SUBSCRIBING, awaitItem())
+                        assertEquals(RealtimeChannel.Status.SUBSCRIBED, awaitItem())
+                        val job = channel.presenceChangeFlow().launchIn(this@runTest)
+                        assertEquals(RealtimeChannel.Status.UNSUBSCRIBING, awaitItem())
+                        assertEquals(RealtimeChannel.Status.UNSUBSCRIBED, awaitItem())
+                        assertEquals(RealtimeChannel.Status.SUBSCRIBING, awaitItem())
+                        assertEquals(RealtimeChannel.Status.SUBSCRIBED, awaitItem())
+                        job.cancel()
+                    }
+                }
+            )
+        }
+    }
+
     //For more complex tests we need integration tests
 
     private fun RealtimeChannel.flowFromEventType(event: String, schema: String, table: String, filter: FilterOperation): Flow<PostgresAction> {

Realtime/src/commonTest/kotlin/RealtimeExtTest.kt

@@ -44,7 +44,7 @@ class RealtimeExtTest {
             )
             createTestClient(
                 wsHandler = { i, o ->
-                    handleSubscribe(i, o, "channelId")
+                    handleSubscribe(i, o, "channelId", true)
                 },
                 supabaseHandler = {
                     val channel = it.channel("channelId")