fix(android): isolate realtime connections by client

templates/android/library/src/main/java/io/package/Client.kt.twig

@@ -11,6 +11,7 @@ import {{ sdk.namespace | caseDot }}.extensions.toJsonRequestBody
 import {{ sdk.namespace | caseDot }}.extensions.toJsonWithNulls
 import {{ sdk.namespace | caseDot }}.models.InputFile
 import {{ sdk.namespace | caseDot }}.models.UploadProgress
+import {{ sdk.namespace | caseDot }}.services.RealtimeConnection
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -74,6 +75,8 @@ class Client @JvmOverloads constructor(
 
     internal lateinit var http: OkHttpClient
 
+    internal val realtime by lazy { RealtimeConnection(this) }
+
     internal val headers: MutableMap<String, String>
 
     val config: MutableMap<String, String>

templates/android/library/src/main/java/io/package/services/Realtime.kt.twig

@@ -23,8 +23,93 @@ import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.CoroutineContext
 
+/**
+ * Realtime wrappers share one connection per [Client]. Separate clients keep their
+ * sockets, subscriptions and lifecycle callbacks independent.
+ */
 class Realtime(client: Client) : Service(client), CoroutineScope {
 
+    private val connection = client.realtime
+    private val job = Job()
+
+    // Cancelling one wrapper must not cancel the client's shared connection scope.
+    override val coroutineContext: CoroutineContext
+        get() = Dispatchers.Main + job
+
+    fun onError(callback: (error: Throwable?, statusCode: Int?) -> Unit) {
+        connection.onError(callback)
+    }
+
+    fun onClose(callback: () -> Unit) {
+        connection.onClose(callback)
+    }
+
+    fun onOpen(callback: () -> Unit) {
+        connection.onOpen(callback)
+    }
+
+    /** Close this client's connection and remove all of its subscriptions and callbacks. */
+    fun disconnect() {
+        connection.disconnect()
+    }
+
+    /**
+     * Fire-and-forget presence upsert. Records the latest payload in state so
+     * that — if the WebSocket isn't open yet, or later reconnects — the most
+     * recent presence is automatically (re)sent on the next `connected` event.
+     * When the socket is already open, the frame is sent immediately.
+     *
+     * @param status Presence status (required).
+     * @param presenceId Presence ID (required).
+     * @param permissions Optional permission list to attach to the presence document.
+     * @param metadata Optional metadata payload.
+     */
+    fun upsertPresence(
+        status: String,
+        presenceId: String,
+        permissions: List<String>? = null,
+        metadata: Map<String, Any>? = null,
+    ) {
+        connection.upsertPresence(status, presenceId, permissions, metadata)
+    }
+
+    fun subscribe(
+        vararg channels: Channel<*>,
+        callback: (RealtimeResponseEvent<Any>) -> Unit,
+    ) = connection.subscribe(*channels, callback = callback)
+
+    fun subscribe(
+        vararg channels: String,
+        callback: (RealtimeResponseEvent<Any>) -> Unit,
+    ) = connection.subscribe(*channels, callback = callback)
+
+    fun <T> subscribe(
+        vararg channels: Channel<*>,
+        payloadType: Class<T>,
+        queries: Set<String> = emptySet(),
+        callback: (RealtimeResponseEvent<T>) -> Unit,
+    ): RealtimeSubscription = connection.subscribe(
+        channels = channels,
+        payloadType = payloadType,
+        queries = queries,
+        callback = callback
+    )
+
+    fun <T> subscribe(
+        vararg channels: String,
+        payloadType: Class<T>,
+        queries: Set<String> = emptySet(),
+        callback: (RealtimeResponseEvent<T>) -> Unit,
+    ): RealtimeSubscription = connection.subscribe(
+        channels = channels,
+        payloadType = payloadType,
+        queries = queries,
+        callback = callback
+    )
+}
+
+internal class RealtimeConnection(private val client: Client) : CoroutineScope {
+
     private val job = Job()
 
     override val coroutineContext: CoroutineContext
@@ -36,25 +121,25 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
         private const val TYPE_EVENT = "event"
         private const val TYPE_PONG = "pong"
         private const val HEARTBEAT_INTERVAL = 20_000L // 20 seconds
+    }
 
-        private var socket: WebSocket? = null
-        private val activeSubscriptions = ConcurrentHashMap<String, RealtimeCallback>()
-        private val pendingSubscribes = LinkedHashMap<String, Map<String, Any>>()
-        private var pendingPresence: Map<String, Any>? = null
-        @Volatile private var appConnected = false
+    private var socket: WebSocket? = null
+    private val activeSubscriptions = ConcurrentHashMap<String, RealtimeCallback>()
+    private val pendingSubscribes = LinkedHashMap<String, Map<String, Any>>()
+    private var pendingPresence: Map<String, Any>? = null
+    @Volatile private var appConnected = false
 
-        private var reconnectAttempts = 0
-        private val socketGeneration = AtomicInteger(0)
-        private var reconnect = true
-        private var heartbeatJob: Job? = null
+    private var reconnectAttempts = 0
+    private val socketGeneration = AtomicInteger(0)
+    private var reconnect = true
+    private var heartbeatJob: Job? = null
 
-        private val subscriptionLock = Any()
-        private val presenceLock = Any()
+    private val subscriptionLock = Any()
+    private val presenceLock = Any()
 
-        private val onErrorCallbacks = CopyOnWriteArrayList<(Throwable?, Int?) -> Unit>()
-        private val onCloseCallbacks = CopyOnWriteArrayList<() -> Unit>()
-        private val onOpenCallbacks = CopyOnWriteArrayList<() -> Unit>()
-    }
+    private val onErrorCallbacks = CopyOnWriteArrayList<(Throwable?, Int?) -> Unit>()
+    private val onCloseCallbacks = CopyOnWriteArrayList<() -> Unit>()
+    private val onOpenCallbacks = CopyOnWriteArrayList<() -> Unit>()
 
     fun onError(callback: (error: Throwable?, statusCode: Int?) -> Unit) {
         onErrorCallbacks.add(callback)
@@ -103,7 +188,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             val queryParams = "project=$encodedProject"
             val url = "${client.endpointRealtime}/realtime?$queryParams"
             val requestBuilder = Request.Builder().url(url)
-            val jwt = client.config["jwt"]
+            val jwt = client.headers["x-{{ spec.info.title | caseLower }}-jwt"]
             if (!jwt.isNullOrEmpty()) {
                 requestBuilder.addHeader("x-{{ spec.info.title | caseLower }}-jwt", jwt)
             }
@@ -233,17 +318,6 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
         }
     }
 
-    /**
-     * Fire-and-forget presence upsert. Records the latest payload in state so
-     * that — if the WebSocket isn't open yet, or later reconnects — the most
-     * recent presence is automatically (re)sent on the next `connected` event.
-     * When the socket is already open, the frame is sent immediately.
-     *
-     * @param status Presence status (required).
-     * @param presenceId Presence ID (required).
-     * @param permissions Optional permission list to attach to the presence document.
-     * @param metadata Optional metadata payload.
-     */
     fun upsertPresence(
         status: String,
         presenceId: String,
@@ -498,7 +572,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             if (isStale(webSocket)) return
             appConnected = false
             stopHeartbeat()
-            // Release the companion ref so createSocket() (called below or by
+            // Release this client's socket ref so createSocket() (called below or by
             // a concurrent subscribe()/upsertPresence()) can open a new one.
             synchronized(subscriptionLock) {
                 if (socket === webSocket) socket = null
@@ -512,7 +586,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             val timeout = getTimeout()
 
             Log.e(
-                this@Realtime::class.java.name,
+                this@RealtimeConnection::class.java.name,
                 "Realtime disconnected. Re-connecting in ${timeout / 1000} seconds.",
                 {{ spec.info.title | caseUcfirst }}Exception(reason, code)
             )

tests/e2e/Android16Java17Test.php

@@ -26,6 +26,8 @@ final class Android16Java17Test extends Base
     protected array $build = [
         'mkdir -p tests/e2e/sdks/android/library/src/test/java',
         'cp tests/e2e/languages/android/Tests.kt tests/e2e/sdks/android/library/src/test/java/Tests.kt',
+        'cp tests/e2e/languages/android/RealtimeIsolationTest.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolationTest.kt',
+        'cat tests/e2e/languages/android/realtime.gradle.kts >> tests/e2e/sdks/android/library/build.gradle.kts',
         'mkdir -p tests/e2e/sdks/android/library/src/test/java/io/appwrite',
         'cp tests/e2e/languages/kotlin/JavaInterop.java tests/e2e/sdks/android/library/src/test/java/io/appwrite/JavaInterop.java',
         'cp tests/e2e/languages/android/ChannelInterop.java tests/e2e/sdks/android/library/src/test/java/io/appwrite/ChannelInterop.java',

tests/e2e/Android5Java17Test.php

@@ -26,6 +26,8 @@ final class Android5Java17Test extends Base
     protected array $build = [
         'mkdir -p tests/e2e/sdks/android/library/src/test/java',
         'cp tests/e2e/languages/android/Tests.kt tests/e2e/sdks/android/library/src/test/java/Tests.kt',
+        'cp tests/e2e/languages/android/RealtimeIsolationTest.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolationTest.kt',
+        'cat tests/e2e/languages/android/realtime.gradle.kts >> tests/e2e/sdks/android/library/build.gradle.kts',
         'mkdir -p tests/e2e/sdks/android/library/src/test/java/io/appwrite',
         'cp tests/e2e/languages/kotlin/JavaInterop.java tests/e2e/sdks/android/library/src/test/java/io/appwrite/JavaInterop.java',
         'cp tests/e2e/languages/android/ChannelInterop.java tests/e2e/sdks/android/library/src/test/java/io/appwrite/ChannelInterop.java',