Realtime exception handler fix

mock-server/src/Utopia/Realtime/Protocol.php

@@ -105,6 +105,16 @@ private function handleSubscribe(Server $server, Connection $connection, mixed $
             }
             $connection->subscribe($subscriptionId, $channels, $queries);
 
+            if (in_array('error', $channels, true)) {
+                $this->error(
+                    $server,
+                    $connection->fd,
+                    'Team is marked as read-only, so you cannot invite new members.',
+                    1008
+                );
+                continue;
+            }
+
             $eventPayload = ['response' => 'WS:/v1/realtime:passed'];
             if (!$this->subscriptionMatchesPayload($queries, $eventPayload)) {
                 continue;

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

@@ -19,6 +19,7 @@ import okhttp3.Response
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.coroutines.CoroutineContext
 
@@ -49,6 +50,34 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
 
         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>()
+    }
+
+    fun onError(callback: (error: Throwable?, statusCode: Int?) -> Unit) {
+        onErrorCallbacks.add(callback)
+    }
+
+    fun onClose(callback: () -> Unit) {
+        onCloseCallbacks.add(callback)
+    }
+
+    fun onOpen(callback: () -> Unit) {
+        onOpenCallbacks.add(callback)
+    }
+
+    private fun dispatchError(error: Throwable?, statusCode: Int?) {
+        onErrorCallbacks.forEach { runCatching { it(error, statusCode) } }
+    }
+
+    private fun dispatchClose() {
+        onCloseCallbacks.forEach { runCatching { it() } }
+    }
+
+    private fun dispatchOpen() {
+        onOpenCallbacks.forEach { runCatching { it() } }
     }
 
     private fun createSocket() {
@@ -144,6 +173,9 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             pendingPresence = null
         }
         appConnected = false
+        onErrorCallbacks.clear()
+        onCloseCallbacks.clear()
+        onOpenCallbacks.clear()
     }
 
     private fun sendPendingSubscribes() {
@@ -390,6 +422,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             if (isStale(webSocket)) return
             reconnectAttempts = 0
             startHeartbeat()
+            dispatchOpen()
         }
 
         override fun onMessage(webSocket: WebSocket, text: String) {
@@ -398,12 +431,17 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
 
             launch(IO) {
                 if (isStale(webSocket)) return@launch
-                val message = text.fromJson<RealtimeResponse>()
-                when (message.type) {
-                    TYPE_ERROR -> handleResponseError(message)
-                    TYPE_CONNECTED -> handleResponseConnected(message)
-                    TYPE_EVENT -> handleResponseEvent(message)
-                    TYPE_PONG -> {}
+                try {
+                    val message = text.fromJson<RealtimeResponse>()
+                    when (message.type) {
+                        TYPE_ERROR -> handleResponseError(message)
+                        TYPE_CONNECTED -> handleResponseConnected(message)
+                        TYPE_EVENT -> handleResponseEvent(message)
+                        TYPE_PONG -> {}
+                    }
+                } catch (e: Throwable) {
+                    val statusCode = (e as? {{ spec.info.title | caseUcfirst }}Exception)?.code
+                    dispatchError(e, statusCode)
                 }
             }
         }
@@ -459,6 +497,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             synchronized(subscriptionLock) {
                 if (socket === webSocket) socket = null
             }
+            dispatchClose()
             if (!reconnect || code == RealtimeCode.POLICY_VIOLATION.value) {
                 reconnect = true
                 return
@@ -490,7 +529,7 @@ class Realtime(client: Client) : Service(client), CoroutineScope {
             synchronized(subscriptionLock) {
                 if (socket === webSocket) socket = null
             }
-            t.printStackTrace()
+            dispatchError(t, null)
         }
     }
 }

templates/flutter/lib/src/realtime_mixin.dart.twig

@@ -382,7 +382,11 @@ mixin RealtimeMixin {
 
   void handleError(RealtimeResponse response) {
     if (response.data['code'] == status.policyViolation) {
-      throw {{spec.info.title | caseUcfirst}}Exception(response.data["message"], response.data["code"]);
+      final error = {{spec.info.title | caseUcfirst}}Exception(
+          response.data["message"], response.data["code"]);
+      for (var subscription in _subscriptions.values) {
+        subscription.controller.addError(error);
+      }
     } else {
       _retry();
     }

tests/e2e/Base.php

@@ -126,6 +126,7 @@ abstract class Base extends TestCase
         'WS:/v1/realtime:passed',
         'WS:/v1/realtime:passed',
         'Realtime failed!',
+        'Realtime error:passed',
         'Realtime unsubscribe:passed',
         'Realtime update:passed',
         'Realtime presence:passed',

tests/e2e/languages/android/Tests.kt

@@ -138,6 +138,12 @@ class ServiceTest {
             }
         }
 
+        var realtimeErrorResponse = "Realtime error:failed"
+        realtime.onError { _, _ ->
+            realtimeErrorResponse = "Realtime error:passed"
+        }
+        realtime.subscribe("error", payloadType = Any::class.java) { }
+
         runBlocking {
             var mock: Mock
 
@@ -252,6 +258,7 @@ class ServiceTest {
             writeToFile(realtimeResponse)
             writeToFile(realtimeResponseWithQueries)
             writeToFile(realtimeResponseWithQueriesFailure)
+            writeToFile(realtimeErrorResponse)
 
             try {
                 rtsubWithQueriesFailure.unsubscribe()

tests/e2e/languages/apple/Tests.swift

@@ -90,6 +90,14 @@ class Tests: XCTestCase {
             print("Realtime presence:passed")
         }
 
+        var realtimeErrorResponse = "Realtime error:failed"
+        let expectationError = XCTestExpectation(description: "realtime error")
+        realtime.onError { _, _ in
+            realtimeErrorResponse = "Realtime error:passed"
+            expectationError.fulfill()
+        }
+        _ = try await realtime.subscribe(channels: ["error"]) { _ in }
+
         var mock: Mock
 
         // Foo Tests
@@ -218,6 +226,9 @@ class Tests: XCTestCase {
             print("Realtime failed")
         }
 
+        await fulfillment(of: [expectationError], timeout: 20.0)
+        print(realtimeErrorResponse)
+
         do {
             try await rtsubWithQueriesFailure.unsubscribe()
             rtsubFailureUnsubscribed = true

tests/e2e/languages/flutter/tests.dart

@@ -218,6 +218,23 @@ void main() async {
     print("Realtime failed!");
   }
 
+  final errorRealtime = Realtime(client);
+  final rtError = Completer<void>();
+  final errorSub = errorRealtime.subscribe(["error"]);
+  errorSub.stream.listen(
+    (_) {},
+    onError: (e) {
+      if (!rtError.isCompleted) rtError.complete();
+    },
+  );
+  try {
+    await rtError.future.timeout(Duration(seconds: 10));
+    print("Realtime error:passed");
+  } catch (e) {
+    print("Realtime error:failed");
+  }
+  await errorRealtime.disconnect();
+
   try {
     await rtsubWithQueriesFailure.unsubscribe();
 

tests/e2e/languages/react-native/browser.js

@@ -70,6 +70,10 @@ import {
         }
     });
 
+    let responseRealtimeError = 'Realtime error:failed';
+    realtime.onError(() => { responseRealtimeError = 'Realtime error:passed'; });
+    const rtsubError = await realtime.subscribe(['error'], () => {});
+
     // Foo
     response = await foo.get('string', 123, ['string in array']);
     console.log(response.result);
@@ -169,6 +173,7 @@ import {
     console.log(responseRealtime);
     console.log(responseRealtimeWithQueries);
     console.log(responseRealtimeWithQueriesFailure);
+    console.log(responseRealtimeError);
 
     try {
         await rtsubWithQueriesFailure.unsubscribe();

tests/e2e/languages/unity/Tests.cs

@@ -104,6 +104,13 @@ private async Task RunAsyncTest()
                 }
             });
 
+            var realtimeErrorResponse = "Realtime error:failed";
+            realtime.OnError += (error) =>
+            {
+                realtimeErrorResponse = "Realtime error:passed";
+            };
+            realtime.Subscribe(new[] { "error" }, (eventData) => { });
+
             await Task.Delay(5000);
 
             // Ping test
@@ -232,6 +239,7 @@ private async Task RunAsyncTest()
             LogResult(realtimeResponse);
             LogResult(realtimeResponseWithQueries);
             LogResult(realtimeResponseWithQueriesFailure);
+            LogResult(realtimeErrorResponse);
 
             try
             {

tests/e2e/languages/web/index.html

@@ -83,6 +83,10 @@
                 }
             });
 
+            let responseRealtimeError = 'Realtime error:failed';
+            realtime.onError(() => { responseRealtimeError = 'Realtime error:passed'; });
+            const rtsubError = await realtime.subscribe(['error'], () => {});
+
             // Foo
             response = await foo.get('string', 123, ["string in array"]);
             console.log(response.result);
@@ -294,6 +298,7 @@
             console.log(responseRealtime)
             console.log(responseRealtimeWithQueries)
             console.log(responseRealtimeWithQueriesFailure)
+            console.log(responseRealtimeError)
 
             try {
                 await rtsubWithQueriesFailure.unsubscribe();