From 3415f1d987b7c3dae4655274f0f9510b459b0ada Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 23 Oct 2026 11:43:08 +0100 Subject: [PATCH 1/4] fix(android): isolate realtime connections by client --- .../src/main/java/io/package/Client.kt.twig | 3 + .../java/io/package/services/Realtime.kt.twig | 128 ++++++++--- tests/e2e/Android16Java17Test.php | 3 + tests/e2e/Android5Java17Test.php | 3 + .../languages/android/RealtimeIsolation.kt | 209 ++++++++++++++++++ tests/e2e/languages/android/Tests.kt | 3 + .../e2e/languages/android/realtime.gradle.kts | 3 + 7 files changed, 324 insertions(+), 28 deletions(-) create mode 100644 tests/e2e/languages/android/RealtimeIsolation.kt create mode 100644 tests/e2e/languages/android/realtime.gradle.kts diff --git a/templates/android/library/src/main/java/io/package/Client.kt.twig b/templates/android/library/src/main/java/io/package/Client.kt.twig index 81c048f6bb..a223fce581 100644 --- a/templates/android/library/src/main/java/io/package/Client.kt.twig +++ b/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 val config: MutableMap diff --git a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig index 256fb55354..5058847f22 100644 --- a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig +++ b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig @@ -23,8 +23,91 @@ 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 + + override val coroutineContext: CoroutineContext + get() = connection.coroutineContext + + 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? = null, + metadata: Map? = null, + ) { + connection.upsertPresence(status, presenceId, permissions, metadata) + } + + fun subscribe( + vararg channels: Channel<*>, + callback: (RealtimeResponseEvent) -> Unit, + ) = connection.subscribe(*channels, callback = callback) + + fun subscribe( + vararg channels: String, + callback: (RealtimeResponseEvent) -> Unit, + ) = connection.subscribe(*channels, callback = callback) + + fun subscribe( + vararg channels: Channel<*>, + payloadType: Class, + queries: Set = emptySet(), + callback: (RealtimeResponseEvent) -> Unit, + ): RealtimeSubscription = connection.subscribe( + channels = channels, + payloadType = payloadType, + queries = queries, + callback = callback + ) + + fun subscribe( + vararg channels: String, + payloadType: Class, + queries: Set = emptySet(), + callback: (RealtimeResponseEvent) -> 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 +119,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() - private val pendingSubscribes = LinkedHashMap>() - private var pendingPresence: Map? = null - @Volatile private var appConnected = false + private var socket: WebSocket? = null + private val activeSubscriptions = ConcurrentHashMap() + private val pendingSubscribes = LinkedHashMap>() + private var pendingPresence: Map? = 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) @@ -233,17 +316,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 +570,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 +584,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) ) diff --git a/tests/e2e/Android16Java17Test.php b/tests/e2e/Android16Java17Test.php index 5e9cd3ee12..0a4e6a8fb9 100644 --- a/tests/e2e/Android16Java17Test.php +++ b/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/RealtimeIsolation.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolation.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', @@ -62,6 +64,7 @@ final class Android16Java17Test extends Base ...Base::ENUM_ARRAY_MODEL_RESPONSES, ...Base::EXCEPTION_RESPONSES, ...Base::REALTIME_RESPONSES, + 'Realtime client isolation:passed', // ...Base::COOKIE_RESPONSES, ...Base::QUERY_HELPER_RESPONSES, Base::QUERY_TRANSPORT_RESPONSE, diff --git a/tests/e2e/Android5Java17Test.php b/tests/e2e/Android5Java17Test.php index fc5f6081d2..783ade5763 100644 --- a/tests/e2e/Android5Java17Test.php +++ b/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/RealtimeIsolation.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolation.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', @@ -59,6 +61,7 @@ final class Android5Java17Test extends Base ...Base::ENUM_ARRAY_MODEL_RESPONSES, ...Base::EXCEPTION_RESPONSES, ...Base::REALTIME_RESPONSES, + 'Realtime client isolation:passed', // ...Base::COOKIE_RESPONSES, ...Base::QUERY_HELPER_RESPONSES, Base::QUERY_TRANSPORT_RESPONSE, diff --git a/tests/e2e/languages/android/RealtimeIsolation.kt b/tests/e2e/languages/android/RealtimeIsolation.kt new file mode 100644 index 0000000000..b4ca73932c --- /dev/null +++ b/tests/e2e/languages/android/RealtimeIsolation.kt @@ -0,0 +1,209 @@ +package io.appwrite + +import android.content.Context +import io.appwrite.extensions.fromJson +import io.appwrite.extensions.toJson +import io.appwrite.services.Realtime +import mockwebserver3.MockResponse +import mockwebserver3.MockWebServer +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.CountDownLatch +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference + +data class IsolationPayload(val source: String) + +/** Exercises the public API against independent WebSocket backends. */ +object RealtimeIsolationScenarios { + fun run(context: Context) { + MockWebServer().use { serverA -> + MockWebServer().use { serverB -> + serverA.start() + serverB.start() + val backendA = Backend("a") + val backendB = Backend("b") + val backendC = Backend("c") + val reconnectB = Backend("b-reconnected") + serverA.enqueue(upgrade(backendA)) + serverA.enqueue(upgrade(backendC)) + serverB.enqueue(upgrade(backendB)) + serverB.enqueue(upgrade(reconnectB)) + + val clientA = client(context, serverA, "project-a", "jwt-a") + val clientB = client(context, serverB, "project-b", "jwt-b") + val clientC = client(context, serverA, "project-a", "jwt-c") + val realtimeA = Realtime(clientA) + val realtimeB = Realtime(clientB) + val realtimeC = Realtime(clientC) + val sharedA = Realtime(clientA) + val eventsA = LinkedBlockingQueue() + val eventsB = LinkedBlockingQueue() + val eventsC = LinkedBlockingQueue() + val sharedEventsA = LinkedBlockingQueue() + val errorsA = LinkedBlockingQueue() + val errorsB = LinkedBlockingQueue() + val errorsC = LinkedBlockingQueue() + val closesA = AtomicInteger() + val closesC = AtomicInteger() + val closeB = CountDownLatch(1) + + realtimeA.onError { _, code -> errorsA.offer(code ?: -1) } + realtimeB.onError { _, code -> errorsB.offer(code ?: -1) } + realtimeC.onError { _, code -> errorsC.offer(code ?: -1) } + realtimeA.onClose { closesA.incrementAndGet() } + realtimeB.onClose { closeB.countDown() } + realtimeC.onClose { closesC.incrementAndGet() } + + try { + subscribe(realtimeA, eventsA) + expect(eventsA, "a", "first backend") + assertHandshake(serverA, "project-a", "jwt-a") + + subscribe(realtimeB, eventsB) + expect(eventsB, "b", "second backend must receive its own subscription") + assertHandshake(serverB, "project-b", "jwt-b") + + // The same endpoint and project still need distinct connections for different clients. + subscribe(realtimeC, eventsC) + expect(eventsC, "c", "same endpoint with another client") + assertHandshake(serverA, "project-a", "jwt-c") + + // Wrappers around the same Client retain one shared connection. + subscribe(sharedA, sharedEventsA) + expect(sharedEventsA, "a", "same-client wrapper") + check(serverA.requestCount == 2) { "same Client opened another socket" } + check(serverB.requestCount == 1) { "unexpected second-backend socket" } + + realtimeA.upsertPresence("online", "presence-a") + realtimeB.upsertPresence("online", "presence-b") + realtimeC.upsertPresence("online", "presence-c") + expect(backendA.presences, "presence-a", "first-client presence") + expect(backendB.presences, "presence-b", "second-client presence") + expect(backendC.presences, "presence-c", "same-endpoint client presence") + + backendB.error() + check(errorsB.poll(10, TimeUnit.SECONDS) == 400) { "missing second-client error" } + check(errorsA.isEmpty() && errorsC.isEmpty()) { "error leaked across clients" } + + backendB.closeForReconnect() + check(closeB.await(10, TimeUnit.SECONDS)) { "missing second-client close callback" } + expect(eventsB, "b-reconnected", "reconnect must stay on the second backend") + assertHandshake(serverB, "project-b", "jwt-b") + check(closesA.get() == 0 && closesC.get() == 0) { "close leaked across clients" } + + realtimeB.disconnect() + check(reconnectB.closed.await(10, TimeUnit.SECONDS)) { "second client did not disconnect" } + backendA.emit("a-after-b-disconnect") + expect(eventsA, "a-after-b-disconnect", "first client after second-client disconnect") + expect(sharedEventsA, "a-after-b-disconnect", "same-client subscription remains active") + + realtimeA.disconnect() + check(backendA.closed.await(10, TimeUnit.SECONDS)) { "first client did not disconnect" } + backendC.emit("c-after-a-disconnect") + expect(eventsC, "c-after-a-disconnect", "same-endpoint client after first-client disconnect") + check(eventsB.isEmpty()) { "disconnected client received another client's event" } + } finally { + realtimeA.disconnect() + realtimeB.disconnect() + realtimeC.disconnect() + } + } + } + } + + private fun client(context: Context, server: MockWebServer, project: String, jwt: String): Client = + Client(context) + .setEndpointRealtime(server.url("/v1").toString().replaceFirst("http", "ws")) + .setProject(project) + .setJWT(jwt) + + private fun subscribe(realtime: Realtime, events: LinkedBlockingQueue) { + realtime.subscribe("tests", payloadType = IsolationPayload::class.java) { + events.offer(it.payload.source) + } + } + + private fun expect(events: LinkedBlockingQueue, expected: String, context: String) { + val actual = events.poll(10, TimeUnit.SECONDS) + check(actual == expected) { "$context: expected $expected, got $actual" } + } + + private fun assertHandshake(server: MockWebServer, project: String, jwt: String) { + val request = checkNotNull(server.takeRequest(10, TimeUnit.SECONDS)) { "missing WebSocket handshake" } + check(request.url.queryParameter("project") == project) { "wrong project in handshake" } + check(request.headers["x-appwrite-jwt"] == jwt) { "wrong credentials in handshake" } + } + + private fun upgrade(listener: WebSocketListener): MockResponse = + MockResponse.Builder().webSocketUpgrade(listener).build() + + private class Backend(private val source: String) : WebSocketListener() { + private val socket = AtomicReference() + private val subscriptions = ConcurrentHashMap>() + val closed = CountDownLatch(1) + val presences = LinkedBlockingQueue() + + override fun onOpen(webSocket: WebSocket, response: Response) { + socket.set(webSocket) + webSocket.send(mapOf("type" to "connected", "data" to emptyMap()).toJson()) + } + + override fun onMessage(webSocket: WebSocket, text: String) { + val frame = text.fromJson>() + if (frame["type"] == "presence") { + val presence = frame["data"] as Map<*, *> + presences.offer(presence["presenceId"] as String) + return + } + if (frame["type"] != "subscribe") return + val rows = frame["data"] as List<*> + rows.forEach { row -> + val subscription = row as Map<*, *> + val id = subscription["subscriptionId"] as String + val channels = (subscription["channels"] as List<*>).map { it.toString() } + subscriptions[id] = channels + emit(source, id, channels) + } + } + + fun emit(source: String) { + subscriptions.forEach { (id, channels) -> emit(source, id, channels) } + } + + private fun emit(source: String, id: String, channels: List) { + check(socket.get().send( + mapOf( + "type" to "event", + "data" to mapOf( + "events" to listOf("tests.update"), + "channels" to channels, + "timestamp" to "2026-01-01T00:00:00.000Z", + "subscriptions" to listOf(id), + "payload" to mapOf("source" to source), + ), + ).toJson() + )) { "could not emit test event" } + } + + fun error() { + socket.get().send(mapOf("type" to "error", "data" to mapOf("message" to "test error", "code" to 400)).toJson()) + } + + fun closeForReconnect() { + check(socket.get().close(1011, "reconnect this client")) + } + + override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { + webSocket.close(code, reason) + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + closed.countDown() + } + } +} diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 6e9172f430..cacdc30336 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -372,6 +372,9 @@ class ServiceTest { writeToFile("Realtime disconnect:failed") } + RealtimeIsolationScenarios.run(ApplicationProvider.getApplicationContext()) + writeToFile("Realtime client isolation:passed") + // mock = general.setCookie() // writeToFile(mock.result) diff --git a/tests/e2e/languages/android/realtime.gradle.kts b/tests/e2e/languages/android/realtime.gradle.kts new file mode 100644 index 0000000000..6757edffbb --- /dev/null +++ b/tests/e2e/languages/android/realtime.gradle.kts @@ -0,0 +1,3 @@ +dependencies { + testImplementation("com.squareup.okhttp3:mockwebserver3:5.5.1") +} From 0134717aa205eb6bcc9c6d515503ae38538e94c9 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 23 Oct 2026 11:46:34 +0100 Subject: [PATCH 2/4] fix(android): keep wrapper cancellation scopes independent --- .../java/io/package/services/Realtime.kt.twig | 4 +- .../languages/android/RealtimeIsolation.kt | 103 +++++++++++++----- tests/e2e/languages/android/r8-rules.pro | 1 + 3 files changed, 82 insertions(+), 26 deletions(-) diff --git a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig index 5058847f22..e4b881a68d 100644 --- a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig +++ b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig @@ -30,9 +30,11 @@ import kotlin.coroutines.CoroutineContext 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() = connection.coroutineContext + get() = Dispatchers.Main + job fun onError(callback: (error: Throwable?, statusCode: Int?) -> Unit) { connection.onError(callback) diff --git a/tests/e2e/languages/android/RealtimeIsolation.kt b/tests/e2e/languages/android/RealtimeIsolation.kt index b4ca73932c..a94e852a15 100644 --- a/tests/e2e/languages/android/RealtimeIsolation.kt +++ b/tests/e2e/languages/android/RealtimeIsolation.kt @@ -4,6 +4,7 @@ import android.content.Context import io.appwrite.extensions.fromJson import io.appwrite.extensions.toJson import io.appwrite.services.Realtime +import kotlinx.coroutines.cancel import mockwebserver3.MockResponse import mockwebserver3.MockWebServer import okhttp3.Response @@ -16,7 +17,9 @@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicReference -data class IsolationPayload(val source: String) +data class IsolationPayload( + val source: String, +) /** Exercises the public API against independent WebSocket backends. */ object RealtimeIsolationScenarios { @@ -51,6 +54,7 @@ object RealtimeIsolationScenarios { val closesA = AtomicInteger() val closesC = AtomicInteger() val closeB = CountDownLatch(1) + var recreatedA: Realtime? = null realtimeA.onError { _, code -> errorsA.offer(code ?: -1) } realtimeB.onError { _, code -> errorsB.offer(code ?: -1) } @@ -102,12 +106,23 @@ object RealtimeIsolationScenarios { expect(eventsA, "a-after-b-disconnect", "first client after second-client disconnect") expect(sharedEventsA, "a-after-b-disconnect", "same-client subscription remains active") + realtimeA.cancel() realtimeA.disconnect() check(backendA.closed.await(10, TimeUnit.SECONDS)) { "first client did not disconnect" } backendC.emit("c-after-a-disconnect") expect(eventsC, "c-after-a-disconnect", "same-endpoint client after first-client disconnect") check(eventsB.isEmpty()) { "disconnected client received another client's event" } + + // Cancelling a public wrapper must not poison later wrappers for its Client. + serverA.enqueue(upgrade(Backend("a-recreated"))) + val replacement = Realtime(clientA) + recreatedA = replacement + val replacementEvents = LinkedBlockingQueue() + subscribe(replacement, replacementEvents) + expect(replacementEvents, "a-recreated", "new wrapper after cancellation and disconnect") + assertHandshake(serverA, "project-a", "jwt-a") } finally { + recreatedA?.disconnect() realtimeA.disconnect() realtimeB.disconnect() realtimeC.disconnect() @@ -116,44 +131,67 @@ object RealtimeIsolationScenarios { } } - private fun client(context: Context, server: MockWebServer, project: String, jwt: String): Client = + private fun client( + context: Context, + server: MockWebServer, + project: String, + jwt: String, + ): Client = Client(context) .setEndpointRealtime(server.url("/v1").toString().replaceFirst("http", "ws")) .setProject(project) .setJWT(jwt) - private fun subscribe(realtime: Realtime, events: LinkedBlockingQueue) { + private fun subscribe( + realtime: Realtime, + events: LinkedBlockingQueue, + ) { realtime.subscribe("tests", payloadType = IsolationPayload::class.java) { events.offer(it.payload.source) } } - private fun expect(events: LinkedBlockingQueue, expected: String, context: String) { + private fun expect( + events: LinkedBlockingQueue, + expected: String, + context: String, + ) { val actual = events.poll(10, TimeUnit.SECONDS) check(actual == expected) { "$context: expected $expected, got $actual" } } - private fun assertHandshake(server: MockWebServer, project: String, jwt: String) { + private fun assertHandshake( + server: MockWebServer, + project: String, + jwt: String, + ) { val request = checkNotNull(server.takeRequest(10, TimeUnit.SECONDS)) { "missing WebSocket handshake" } check(request.url.queryParameter("project") == project) { "wrong project in handshake" } check(request.headers["x-appwrite-jwt"] == jwt) { "wrong credentials in handshake" } } - private fun upgrade(listener: WebSocketListener): MockResponse = - MockResponse.Builder().webSocketUpgrade(listener).build() + private fun upgrade(listener: WebSocketListener): MockResponse = MockResponse.Builder().webSocketUpgrade(listener).build() - private class Backend(private val source: String) : WebSocketListener() { + private class Backend( + private val source: String, + ) : WebSocketListener() { private val socket = AtomicReference() private val subscriptions = ConcurrentHashMap>() val closed = CountDownLatch(1) val presences = LinkedBlockingQueue() - override fun onOpen(webSocket: WebSocket, response: Response) { + override fun onOpen( + webSocket: WebSocket, + response: Response, + ) { socket.set(webSocket) webSocket.send(mapOf("type" to "connected", "data" to emptyMap()).toJson()) } - override fun onMessage(webSocket: WebSocket, text: String) { + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { val frame = text.fromJson>() if (frame["type"] == "presence") { val presence = frame["data"] as Map<*, *> @@ -175,19 +213,26 @@ object RealtimeIsolationScenarios { subscriptions.forEach { (id, channels) -> emit(source, id, channels) } } - private fun emit(source: String, id: String, channels: List) { - check(socket.get().send( - mapOf( - "type" to "event", - "data" to mapOf( - "events" to listOf("tests.update"), - "channels" to channels, - "timestamp" to "2026-01-01T00:00:00.000Z", - "subscriptions" to listOf(id), - "payload" to mapOf("source" to source), - ), - ).toJson() - )) { "could not emit test event" } + private fun emit( + source: String, + id: String, + channels: List, + ) { + check( + socket.get().send( + mapOf( + "type" to "event", + "data" to + mapOf( + "events" to listOf("tests.update"), + "channels" to channels, + "timestamp" to "2026-01-01T00:00:00.000Z", + "subscriptions" to listOf(id), + "payload" to mapOf("source" to source), + ), + ).toJson(), + ), + ) { "could not emit test event" } } fun error() { @@ -198,11 +243,19 @@ object RealtimeIsolationScenarios { check(socket.get().close(1011, "reconnect this client")) } - override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { + override fun onClosing( + webSocket: WebSocket, + code: Int, + reason: String, + ) { webSocket.close(code, reason) } - override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + override fun onClosed( + webSocket: WebSocket, + code: Int, + reason: String, + ) { closed.countDown() } } diff --git a/tests/e2e/languages/android/r8-rules.pro b/tests/e2e/languages/android/r8-rules.pro index 455fe90646..ff8dc81b0a 100644 --- a/tests/e2e/languages/android/r8-rules.pro +++ b/tests/e2e/languages/android/r8-rules.pro @@ -3,5 +3,6 @@ -keep class <1> { *; } -keep class io.appwrite.E2EPushReceiver { (); } -keep class io.appwrite.TestPayload { *; } +-keep class io.appwrite.IsolationPayload { *; } -dontwarn android.content.pm.** -dontwarn android.app.IActivityManager** From 59ff37ef69b424506548dc05fadfe3a48b8dc103 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 23 Oct 2026 11:50:24 +0100 Subject: [PATCH 3/4] fix(android): use the configured JWT for realtime handshakes --- .../library/src/main/java/io/package/services/Realtime.kt.twig | 2 +- tests/e2e/languages/android/RealtimeIsolation.kt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig index e4b881a68d..b01f7ce060 100644 --- a/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig +++ b/templates/android/library/src/main/java/io/package/services/Realtime.kt.twig @@ -188,7 +188,7 @@ internal class RealtimeConnection(private val client: 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) } diff --git a/tests/e2e/languages/android/RealtimeIsolation.kt b/tests/e2e/languages/android/RealtimeIsolation.kt index a94e852a15..99490004b6 100644 --- a/tests/e2e/languages/android/RealtimeIsolation.kt +++ b/tests/e2e/languages/android/RealtimeIsolation.kt @@ -66,10 +66,10 @@ object RealtimeIsolationScenarios { try { subscribe(realtimeA, eventsA) expect(eventsA, "a", "first backend") - assertHandshake(serverA, "project-a", "jwt-a") subscribe(realtimeB, eventsB) expect(eventsB, "b", "second backend must receive its own subscription") + assertHandshake(serverA, "project-a", "jwt-a") assertHandshake(serverB, "project-b", "jwt-b") // The same endpoint and project still need distinct connections for different clients. From 199403041b60fc0cb10cec3acbd133a50fd12c30 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 23 Oct 2026 12:05:44 +0100 Subject: [PATCH 4/4] test(android): assert realtime isolation directly in JUnit --- tests/e2e/Android16Java17Test.php | 3 +- tests/e2e/Android5Java17Test.php | 3 +- ...eIsolation.kt => RealtimeIsolationTest.kt} | 68 ++++++++++++++----- tests/e2e/languages/android/Tests.kt | 3 - 4 files changed, 52 insertions(+), 25 deletions(-) rename tests/e2e/languages/android/{RealtimeIsolation.kt => RealtimeIsolationTest.kt} (79%) diff --git a/tests/e2e/Android16Java17Test.php b/tests/e2e/Android16Java17Test.php index 0a4e6a8fb9..a9fad34f6a 100644 --- a/tests/e2e/Android16Java17Test.php +++ b/tests/e2e/Android16Java17Test.php @@ -26,7 +26,7 @@ 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/RealtimeIsolation.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolation.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', @@ -64,7 +64,6 @@ final class Android16Java17Test extends Base ...Base::ENUM_ARRAY_MODEL_RESPONSES, ...Base::EXCEPTION_RESPONSES, ...Base::REALTIME_RESPONSES, - 'Realtime client isolation:passed', // ...Base::COOKIE_RESPONSES, ...Base::QUERY_HELPER_RESPONSES, Base::QUERY_TRANSPORT_RESPONSE, diff --git a/tests/e2e/Android5Java17Test.php b/tests/e2e/Android5Java17Test.php index 783ade5763..5a06ce03bb 100644 --- a/tests/e2e/Android5Java17Test.php +++ b/tests/e2e/Android5Java17Test.php @@ -26,7 +26,7 @@ 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/RealtimeIsolation.kt tests/e2e/sdks/android/library/src/test/java/RealtimeIsolation.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', @@ -61,7 +61,6 @@ final class Android5Java17Test extends Base ...Base::ENUM_ARRAY_MODEL_RESPONSES, ...Base::EXCEPTION_RESPONSES, ...Base::REALTIME_RESPONSES, - 'Realtime client isolation:passed', // ...Base::COOKIE_RESPONSES, ...Base::QUERY_HELPER_RESPONSES, Base::QUERY_TRANSPORT_RESPONSE, diff --git a/tests/e2e/languages/android/RealtimeIsolation.kt b/tests/e2e/languages/android/RealtimeIsolationTest.kt similarity index 79% rename from tests/e2e/languages/android/RealtimeIsolation.kt rename to tests/e2e/languages/android/RealtimeIsolationTest.kt index 99490004b6..30db802582 100644 --- a/tests/e2e/languages/android/RealtimeIsolation.kt +++ b/tests/e2e/languages/android/RealtimeIsolationTest.kt @@ -1,15 +1,29 @@ package io.appwrite import android.content.Context +import androidx.test.core.app.ApplicationProvider +import androidx.test.ext.junit.runners.AndroidJUnit4 import io.appwrite.extensions.fromJson import io.appwrite.extensions.toJson import io.appwrite.services.Realtime +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.cancel +import kotlinx.coroutines.test.resetMain +import kotlinx.coroutines.test.setMain import mockwebserver3.MockResponse import mockwebserver3.MockWebServer import okhttp3.Response import okhttp3.WebSocket import okhttp3.WebSocketListener +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.annotation.Config import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CountDownLatch import java.util.concurrent.LinkedBlockingQueue @@ -22,8 +36,23 @@ data class IsolationPayload( ) /** Exercises the public API against independent WebSocket backends. */ -object RealtimeIsolationScenarios { - fun run(context: Context) { +@Config(manifest = Config.NONE) +@RunWith(AndroidJUnit4::class) +@OptIn(ExperimentalCoroutinesApi::class) +class RealtimeIsolationTest { + @Before + fun setUpDispatcher() { + Dispatchers.setMain(Dispatchers.Unconfined) + } + + @After + fun resetDispatcher() { + Dispatchers.resetMain() + } + + @Test + fun isolatesClientConnectionsAndLifecycle() { + val context = ApplicationProvider.getApplicationContext() MockWebServer().use { serverA -> MockWebServer().use { serverB -> serverA.start() @@ -80,8 +109,8 @@ object RealtimeIsolationScenarios { // Wrappers around the same Client retain one shared connection. subscribe(sharedA, sharedEventsA) expect(sharedEventsA, "a", "same-client wrapper") - check(serverA.requestCount == 2) { "same Client opened another socket" } - check(serverB.requestCount == 1) { "unexpected second-backend socket" } + assertEquals("same Client opened another socket", 2, serverA.requestCount) + assertEquals("unexpected second-backend socket", 1, serverB.requestCount) realtimeA.upsertPresence("online", "presence-a") realtimeB.upsertPresence("online", "presence-b") @@ -91,27 +120,28 @@ object RealtimeIsolationScenarios { expect(backendC.presences, "presence-c", "same-endpoint client presence") backendB.error() - check(errorsB.poll(10, TimeUnit.SECONDS) == 400) { "missing second-client error" } - check(errorsA.isEmpty() && errorsC.isEmpty()) { "error leaked across clients" } + assertEquals("missing second-client error", 400, errorsB.poll(10, TimeUnit.SECONDS)) + assertTrue("error leaked across clients", errorsA.isEmpty() && errorsC.isEmpty()) backendB.closeForReconnect() - check(closeB.await(10, TimeUnit.SECONDS)) { "missing second-client close callback" } + assertTrue("missing second-client close callback", closeB.await(10, TimeUnit.SECONDS)) expect(eventsB, "b-reconnected", "reconnect must stay on the second backend") assertHandshake(serverB, "project-b", "jwt-b") - check(closesA.get() == 0 && closesC.get() == 0) { "close leaked across clients" } + assertEquals("close leaked to first client", 0, closesA.get()) + assertEquals("close leaked to same-endpoint client", 0, closesC.get()) realtimeB.disconnect() - check(reconnectB.closed.await(10, TimeUnit.SECONDS)) { "second client did not disconnect" } + assertTrue("second client did not disconnect", reconnectB.closed.await(10, TimeUnit.SECONDS)) backendA.emit("a-after-b-disconnect") expect(eventsA, "a-after-b-disconnect", "first client after second-client disconnect") expect(sharedEventsA, "a-after-b-disconnect", "same-client subscription remains active") realtimeA.cancel() realtimeA.disconnect() - check(backendA.closed.await(10, TimeUnit.SECONDS)) { "first client did not disconnect" } + assertTrue("first client did not disconnect", backendA.closed.await(10, TimeUnit.SECONDS)) backendC.emit("c-after-a-disconnect") expect(eventsC, "c-after-a-disconnect", "same-endpoint client after first-client disconnect") - check(eventsB.isEmpty()) { "disconnected client received another client's event" } + assertTrue("disconnected client received another client's event", eventsB.isEmpty()) // Cancelling a public wrapper must not poison later wrappers for its Client. serverA.enqueue(upgrade(Backend("a-recreated"))) @@ -157,7 +187,7 @@ object RealtimeIsolationScenarios { context: String, ) { val actual = events.poll(10, TimeUnit.SECONDS) - check(actual == expected) { "$context: expected $expected, got $actual" } + assertEquals(context, expected, actual) } private fun assertHandshake( @@ -165,9 +195,10 @@ object RealtimeIsolationScenarios { project: String, jwt: String, ) { - val request = checkNotNull(server.takeRequest(10, TimeUnit.SECONDS)) { "missing WebSocket handshake" } - check(request.url.queryParameter("project") == project) { "wrong project in handshake" } - check(request.headers["x-appwrite-jwt"] == jwt) { "wrong credentials in handshake" } + val request = server.takeRequest(10, TimeUnit.SECONDS) + assertNotNull("missing WebSocket handshake", request) + assertEquals("wrong project in handshake", project, request!!.url.queryParameter("project")) + assertEquals("wrong credentials in handshake", jwt, request.headers["x-appwrite-jwt"]) } private fun upgrade(listener: WebSocketListener): MockResponse = MockResponse.Builder().webSocketUpgrade(listener).build() @@ -218,7 +249,8 @@ object RealtimeIsolationScenarios { id: String, channels: List, ) { - check( + assertTrue( + "could not emit test event", socket.get().send( mapOf( "type" to "event", @@ -232,7 +264,7 @@ object RealtimeIsolationScenarios { ), ).toJson(), ), - ) { "could not emit test event" } + ) } fun error() { @@ -240,7 +272,7 @@ object RealtimeIsolationScenarios { } fun closeForReconnect() { - check(socket.get().close(1011, "reconnect this client")) + assertTrue("could not close test socket", socket.get().close(1011, "reconnect this client")) } override fun onClosing( diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index cacdc30336..6e9172f430 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -372,9 +372,6 @@ class ServiceTest { writeToFile("Realtime disconnect:failed") } - RealtimeIsolationScenarios.run(ApplicationProvider.getApplicationContext()) - writeToFile("Realtime client isolation:passed") - // mock = general.setCookie() // writeToFile(mock.result)