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..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 @@ -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? = 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 +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() - 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) @@ -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) ) diff --git a/tests/e2e/Android16Java17Test.php b/tests/e2e/Android16Java17Test.php index 5e9cd3ee12..a9fad34f6a 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/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', diff --git a/tests/e2e/Android5Java17Test.php b/tests/e2e/Android5Java17Test.php index fc5f6081d2..5a06ce03bb 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/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', diff --git a/tests/e2e/languages/android/RealtimeIsolationTest.kt b/tests/e2e/languages/android/RealtimeIsolationTest.kt new file mode 100644 index 0000000000..30db802582 --- /dev/null +++ b/tests/e2e/languages/android/RealtimeIsolationTest.kt @@ -0,0 +1,294 @@ +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 +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. */ +@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() + 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) + var recreatedA: Realtime? = null + + 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") + + 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. + 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") + 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") + 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() + assertEquals("missing second-client error", 400, errorsB.poll(10, TimeUnit.SECONDS)) + assertTrue("error leaked across clients", errorsA.isEmpty() && errorsC.isEmpty()) + + backendB.closeForReconnect() + 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") + assertEquals("close leaked to first client", 0, closesA.get()) + assertEquals("close leaked to same-endpoint client", 0, closesC.get()) + + realtimeB.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() + 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") + 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"))) + 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() + } + } + } + } + + 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) + assertEquals(context, expected, actual) + } + + private fun assertHandshake( + server: MockWebServer, + project: String, + jwt: String, + ) { + 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() + + 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, + ) { + assertTrue( + "could not emit test event", + 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(), + ), + ) + } + + fun error() { + socket.get().send(mapOf("type" to "error", "data" to mapOf("message" to "test error", "code" to 400)).toJson()) + } + + fun closeForReconnect() { + assertTrue("could not close test socket", 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/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** 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") +}