[ECO-5426] feat: implement handleStateChange method for channel state management

- Added impl. for handling channel attached state
- Updated ObjectSyncTracker along with related unit tests

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/LiveObjectsPlugin.java

@@ -1,5 +1,6 @@
 package io.ably.lib.objects;
 
+import io.ably.lib.realtime.ChannelState;
 import io.ably.lib.types.ProtocolMessage;
 import org.jetbrains.annotations.NotNull;
 
@@ -30,6 +31,17 @@ public interface LiveObjectsPlugin {
      */
     void handle(@NotNull ProtocolMessage message);
 
+    /**
+     * Handles state changes for a specific channel.
+     * This method is invoked whenever a channel's state changes, allowing the implementation
+     * to update the LiveObjects instances accordingly based on the new state and presence of objects.
+     *
+     * @param channelName the name of the channel whose state has changed.
+     * @param state the new state of the channel.
+     * @param hasObjects flag indicates whether the channel has any associated live objects.
+     */
+    void handleStateChange(@NotNull String channelName, @NotNull ChannelState state, boolean hasObjects);
+
     /**
      * Disposes of the LiveObjects instance associated with the specified channel name.
      * This method removes the LiveObjects instance for the given channel, releasing any

lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

@@ -145,6 +145,15 @@ private void setState(ChannelState newState, ErrorInfo reason, boolean resumed,
             this.reason = stateChange.reason;
         }
 
+        // cover states other than attached, ChannelState.attached already covered in setAttached
+        if (liveObjectsPlugin != null && newState!= ChannelState.attached) {
+            try {
+                liveObjectsPlugin.handleStateChange(name, newState, false);
+            } catch (Throwable t) {
+                Log.e(TAG, "Unexpected exception in LiveObjectsPlugin.handle", t);
+            }
+        }
+
         if (newState != ChannelState.attaching && newState != ChannelState.suspended) {
             this.retryAttempt = 0;
         }
@@ -439,6 +448,13 @@ private void setAttached(ProtocolMessage message) {
             }
             return;
         }
+        if (liveObjectsPlugin != null) {
+            try {
+                liveObjectsPlugin.handleStateChange(name, ChannelState.attached, message.hasFlag(Flag.has_objects));
+            } catch (Throwable t) {
+                Log.e(TAG, "Unexpected exception in LiveObjectsPlugin.handle", t);
+            }
+        }
         if(state == ChannelState.attached) {
             Log.v(TAG, String.format(Locale.ROOT, "Server initiated attach for channel %s", name));
             if (!message.hasFlag(Flag.resumed)) { // RTL12

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

@@ -1,5 +1,6 @@
 package io.ably.lib.objects
 
+import io.ably.lib.realtime.ChannelState
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ProtocolMessage
 import io.ably.lib.util.Log
@@ -95,9 +96,9 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
 
   /**
    * Handles a ProtocolMessage containing proto action as `object` or `object_sync`.
-   *  @spec RTL1 - Processes incoming object messages and object sync messages
+   * @spec RTL1 - Processes incoming object messages and object sync messages
    */
-  fun handle(protocolMessage: ProtocolMessage) {
+  internal fun handle(protocolMessage: ProtocolMessage) {
     // RTL15b - Set channel serial for OBJECT messages
     adapter.setChannelSerial(channelName, protocolMessage)
 
@@ -141,6 +142,40 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
     }
   }
 
+  internal fun handleStateChange(state: ChannelState, hasObjects: Boolean) {
+    sequentialScope.launch {
+      when (state) {
+        ChannelState.attached -> {
+          Log.v(tag, "Objects.onAttached() channel=$channelName, hasObjects=$hasObjects")
+
+          // RTO4a
+          val fromInitializedState = this@DefaultLiveObjects.state == ObjectsState.INITIALIZED
+          if (hasObjects || fromInitializedState) {
+            // should always start a new sync sequence if we're in the initialized state, no matter the HAS_OBJECTS flag value.
+            // this guarantees we emit both "syncing" -> "synced" events in that order.
+            objectsManager.startNewSync(null)
+          }
+
+          // RTO4b
+          if (!hasObjects) {
+            // if no HAS_OBJECTS flag received on attach, we can end sync sequence immediately and treat it as no objects on a channel.
+            // reset the objects pool to its initial state, and emit update events so subscribers to root object get notified about changes.
+            objectsPool.resetToInitialPool(true) // RTO4b1, RTO4b2
+            objectsManager.clearSyncObjectsDataPool() // RTO4b3
+            objectsManager.clearBufferedObjectOperations() // RTO4b5
+            // defer the state change event until the next tick if we started a new sequence just now due to being in initialized state.
+            // this allows any event listeners to process the start of the new sequence event that was emitted earlier during this event loop.
+            objectsManager.endSync(fromInitializedState) // RTO4b4
+          }
+        }
+
+        else -> {
+          // No action needed for other states
+        }
+      }
+    }
+  }
+
   /**
    * Changes the state and emits events.
    *

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

@@ -1,5 +1,6 @@
 package io.ably.lib.objects
 
+import io.ably.lib.realtime.ChannelState
 import io.ably.lib.types.ProtocolMessage
 import java.util.concurrent.ConcurrentHashMap
 
@@ -16,6 +17,10 @@ public class DefaultLiveObjectsPlugin(private val adapter: LiveObjectsAdapter) :
     liveObjects[channelName]?.handle(msg)
   }
 
+  override fun handleStateChange(channelName: String, state: ChannelState, hasObjects: Boolean) {
+    liveObjects[channelName]?.handleStateChange(state, hasObjects)
+  }
+
   override fun dispose(channelName: String) {
     liveObjects[channelName]?.dispose()
     liveObjects.remove(channelName)

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

@@ -70,7 +70,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
    *
    * @spec RTO5 - Sync sequence initialization
    */
-  private fun startNewSync(syncId: String?) {
+  internal fun startNewSync(syncId: String?) {
     Log.v(tag, "Starting new sync sequence: syncId=$syncId")
 
     // need to discard all buffered object operation messages on new sync start
@@ -85,7 +85,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
    *
    * @spec RTO5c - Applies sync data and buffered operations
    */
-  private fun endSync(deferStateEvent: Boolean) {
+  internal fun endSync(deferStateEvent: Boolean) {
     Log.v(tag, "Ending sync sequence")
     applySync()
     // should apply buffered object operations after we applied the sync.
@@ -98,6 +98,22 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
     liveObjects.stateChange(ObjectsState.SYNCED, deferStateEvent)
   }
 
+  /**
+   * Clears the sync objects data pool.
+   * Used by DefaultLiveObjects.handleStateChange.
+   */
+  internal fun clearSyncObjectsDataPool() {
+    syncObjectsDataPool.clear()
+  }
+
+  /**
+   * Clears the buffered object operations.
+   * Used by DefaultLiveObjects.handleStateChange.
+   */
+  internal fun clearBufferedObjectOperations() {
+    bufferedObjectOperations.clear()
+  }
+
   /**
    * Applies sync data to objects pool.
    *

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

@@ -63,34 +63,34 @@ internal class ObjectsPool(
   /**
    * Gets a live object from the pool by object ID.
    */
-  fun get(objectId: String): BaseLiveObject? {
+  internal fun get(objectId: String): BaseLiveObject? {
     return pool[objectId]
   }
 
   /**
    * Deletes objects from the pool for which object ids are not found in the provided array of ids.
    */
-  fun deleteExtraObjectIds(objectIds: MutableSet<String>) {
+  internal fun deleteExtraObjectIds(objectIds: MutableSet<String>) {
     pool.entries.removeIf { (key, _) -> key !in objectIds }
   }
 
   /**
    * Sets a live object in the pool.
    */
-  fun set(objectId: String, liveObject: BaseLiveObject) {
+  internal fun set(objectId: String, liveObject: BaseLiveObject) {
     pool[objectId] = liveObject
   }
 
   /**
    * Removes all objects but root from the pool and clears the data for root.
    * Does not create a new root object, so the reference to the root object remains the same.
    */
-  fun resetToInitialPool(emitUpdateEvents: Boolean) {
+  internal fun resetToInitialPool(emitUpdateEvents: Boolean) {
     // Clear the pool first and keep the root object
     val root = pool[ROOT_OBJECT_ID]
     if (root != null) {
       pool.clear()
-      pool[ROOT_OBJECT_ID] = root
+      set(ROOT_OBJECT_ID, root)
 
       // Clear the data, this will only clear the root object
       clearObjectsData(emitUpdateEvents)
@@ -178,33 +178,6 @@ internal class ObjectsPool(
   fun dispose() {
     gcJob?.cancel()
     gcScope.cancel()
-    clear()
+    pool.clear()
   }
-
-  /**
-   * Gets all object IDs in the pool.
-   * Useful for debugging and testing.
-   */
-  fun getObjectIds(): Set<String> = pool.keys.toSet()
-
-  /**
-   * Gets the size of the pool.
-   * Useful for debugging and testing.
-   */
-  fun size(): Int = pool.size
-
-  /**
-   * Checks if the pool contains an object with the given ID.
-   */
-  fun contains(objectId: String): Boolean = pool.containsKey(objectId)
-
-  /**
-   * Removes an object from the pool.
-   */
-  fun remove(objectId: String): BaseLiveObject? = pool.remove(objectId)
-
-  /**
-   * Clears all objects from the pool.
-   */
-  fun clear() = pool.clear()
 }

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

@@ -5,33 +5,36 @@ package io.ably.lib.objects
  */
 internal class ObjectsSyncTracker(syncChannelSerial: String?) {
   internal val syncId: String?
-  private val syncCursor: String?
-  private val hasEnded: Boolean
+  internal val syncCursor: String?
+  private val syncSerial: String? = syncChannelSerial
 
   init {
     val parsed = parseSyncChannelSerial(syncChannelSerial)
     syncId = parsed.first
     syncCursor = parsed.second
-    hasEnded = syncChannelSerial.isNullOrEmpty() || syncCursor.isNullOrEmpty()
   }
 
   /**
    * Checks if a new sync sequence has started.
    *
    * @param prevSyncId The previously stored sync ID
    * @return true if a new sync sequence has started, false otherwise
+   *
+   * Spec: RTO5a5, RTO5a2
    */
   internal fun hasSyncStarted(prevSyncId: String?): Boolean {
-    return prevSyncId != syncId
+    return syncSerial.isNullOrEmpty() || prevSyncId != syncId
   }
 
   /**
    * Checks if the current sync sequence has ended.
    *
    * @return true if the sync sequence has ended, false otherwise
+   *
+   * Spec: RTO5a5, RTO5a4
    */
   internal fun hasSyncEnded(): Boolean {
-    return hasEnded
+    return syncSerial.isNullOrEmpty() || syncCursor.isNullOrEmpty()
   }
 
   companion object {

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

@@ -7,31 +7,37 @@ import org.junit.Assert.*
 class ObjectsSyncTrackerTest {
 
     @Test
-  fun `should parse valid sync channel serial with syncId and cursor`() {
+  fun `(RTO5a, RTO5a1, RTO5a2) Should parse valid sync channel serial with syncId and cursor`() {
     val syncTracker = ObjectsSyncTracker("sync-123:cursor-456")
 
     assertEquals("sync-123", syncTracker.syncId)
     assertFalse(syncTracker.hasSyncStarted("sync-123"))
-
     assertTrue(syncTracker.hasSyncStarted(null))
     assertTrue(syncTracker.hasSyncStarted("sync-124"))
 
+    assertEquals("cursor-456", syncTracker.syncCursor)
     assertFalse(syncTracker.hasSyncEnded())
   }
 
   @Test
-  fun `should handle null sync channel serial`() {
+  fun `(RTO5a5) Should handle null sync channel serial`() {
     val syncTracker = ObjectsSyncTracker(null)
 
     assertNull(syncTracker.syncId)
+    assertTrue(syncTracker.hasSyncStarted(null))
+
+    assertNull(syncTracker.syncCursor)
     assertTrue(syncTracker.hasSyncEnded())
   }
 
   @Test
-  fun `should handle empty sync channel serial`() {
+  fun `(RTO5a5) Should handle empty sync channel serial`() {
     val syncTracker = ObjectsSyncTracker("")
 
     assertNull(syncTracker.syncId)
+    assertTrue(syncTracker.hasSyncStarted(null))
+
+    assertNull(syncTracker.syncCursor)
     assertTrue(syncTracker.hasSyncEnded())
   }
 
@@ -40,21 +46,20 @@ class ObjectsSyncTrackerTest {
     val syncTracker = ObjectsSyncTracker("sync_123-456:cursor_789-012")
 
     assertEquals("sync_123-456", syncTracker.syncId)
-    assertFalse(syncTracker.hasSyncEnded())
-  }
 
-  @Test
-  fun `should detect sync sequence ended when syncChannelSerial is null`() {
-    val syncTracker = ObjectsSyncTracker(null)
-
-    assertTrue(syncTracker.hasSyncEnded())
+    assertEquals("cursor_789-012", syncTracker.syncCursor)
+    assertFalse(syncTracker.hasSyncEnded())
   }
 
   @Test
-  fun `should detect sync sequence ended when sync cursor is empty`() {
+  fun `(RTO5a4) should detect sync sequence ended when sync cursor is empty`() {
     val syncTracker = ObjectsSyncTracker("sync-123:")
+
+    assertEquals("sync-123", syncTracker.syncId)
     assertTrue(syncTracker.hasSyncStarted(null))
     assertTrue(syncTracker.hasSyncStarted(""))
+
+    assertEquals("", syncTracker.syncCursor)
     assertTrue(syncTracker.hasSyncEnded())
   }
 }