[ECO-5426] Refactored initializeHandlerForIncomingObjectMessages to handle exceptions

1. Added try catch to avoid crashing collector
2. Fixed defaultLiveObjects flaky tests

Author: sacOO7

live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjects.kt

@@ -129,14 +129,18 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
             )
           }
 
-        when (protocolMessage.action) {
-          ProtocolMessage.Action.`object` -> objectsManager.handleObjectMessages(objects)
-          ProtocolMessage.Action.object_sync -> objectsManager.handleObjectSyncMessages(
-            objects,
-            protocolMessage.channelSerial
-          )
-
-          else -> Log.w(tag, "Ignoring protocol message with unhandled action: ${protocolMessage.action}")
+        try {
+          when (protocolMessage.action) {
+            ProtocolMessage.Action.`object` -> objectsManager.handleObjectMessages(objects)
+            ProtocolMessage.Action.object_sync -> objectsManager.handleObjectSyncMessages(
+              objects,
+              protocolMessage.channelSerial
+            )
+            else -> Log.w(tag, "Ignoring protocol message with unhandled action: ${protocolMessage.action}")
+          }
+        } catch (exception: Exception) {
+          // Skip current message if an error occurs, don't rethrow to avoid crashing the collector
+          Log.e(tag, "Error handling objects message with protocolMsg id ${protocolMessage.id}", exception)
         }
       }
     }

live-objects/src/test/kotlin/io/ably/lib/objects/unit/objects/DefaultLiveObjectsTest.kt

@@ -34,14 +34,16 @@ class DefaultLiveObjectsTest {
 
     // RTO4a - If the HAS_OBJECTS flag is 1, the server will shortly perform an OBJECT_SYNC sequence
     defaultLiveObjects.handleStateChange(ChannelState.attached, true)
+
+    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCING }
+
     // It is expected that the client will start a new sync sequence
     verify(exactly = 1) {
       defaultLiveObjects.ObjectsManager.startNewSync(null)
     }
     verify(exactly = 0) {
       defaultLiveObjects.ObjectsManager.endSync(any<Boolean>())
     }
-    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCING }
   }
 
   @Test
@@ -57,15 +59,16 @@ class DefaultLiveObjectsTest {
     // RTO4b - If the HAS_OBJECTS flag is 0, the sync sequence must be considered complete immediately
     defaultLiveObjects.handleStateChange(ChannelState.attached, false)
 
+    // Verify expected outcomes
+    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCED } // RTO4b4
+
     verify(exactly = 1) {
       defaultLiveObjects.objectsPool.resetToInitialPool(true)
     }
     verify(exactly = 1) {
       defaultLiveObjects.ObjectsManager.endSync(any<Boolean>())
     }
 
-    // Verify expected outcomes
-    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCED } // RTO4b4
     assertEquals(0, defaultLiveObjects.ObjectsManager.SyncObjectsDataPool.size) // RTO4b3
     assertEquals(0, defaultLiveObjects.ObjectsManager.BufferedObjectOperations.size) // RTO4b5
     assertEquals(1, defaultLiveObjects.objectsPool.size()) // RTO4b1 - Only root remains