[ECO-5426] Added separate managers for handling incoming objectMessages

- Refactored/simplified GC for LiveMapEntries and LiveObjects
- Improved BaseLiveObject interface, added comprehensive doc

Author: sacOO7

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

@@ -1,8 +1,8 @@
 package io.ably.lib.objects
 
-import io.ably.lib.objects.type.*
 import io.ably.lib.objects.type.BaseLiveObject
-import io.ably.lib.objects.type.DefaultLiveCounter
+import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
+import io.ably.lib.objects.type.livemap.DefaultLiveMap
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ProtocolMessage
 import io.ably.lib.util.Log
@@ -246,7 +246,7 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
       // RTO5c1a
       if (existingObject != null) {
         // Update existing object
-        val update = existingObject.overrideWithObjectState(objectState) // RTO5c1a1
+        val update = existingObject.applyObjectState(objectState) // RTO5c1a1
         existingObjectUpdates.add(Pair(existingObject, update))
       } else { // RTO5c1b
         // RTO5c1b1 - Create new object and add it to the pool
@@ -323,7 +323,7 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
       objectState.map != null -> DefaultLiveMap.zeroValue(objectState.objectId, adapter, objectsPool) // RTO5c1b1b
       else -> throw clientError("Object state must contain either counter or map data") // RTO5c1b1c
     }.apply {
-      overrideWithObjectState(objectState)
+      applyObjectState(objectState)
     }
   }
 

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

@@ -1,8 +1,8 @@
 package io.ably.lib.objects
 
 import io.ably.lib.objects.type.BaseLiveObject
-import io.ably.lib.objects.type.DefaultLiveCounter
-import io.ably.lib.objects.type.DefaultLiveMap
+import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
+import io.ably.lib.objects.type.livemap.DefaultLiveMap
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
 import java.util.concurrent.ConcurrentHashMap
@@ -71,9 +71,7 @@ internal class ObjectsPool(
    * Deletes objects from the pool for which object ids are not found in the provided array of ids.
    */
   fun deleteExtraObjectIds(objectIds: MutableSet<String>) {
-    pool.keys.toList()
-      .filter { it !in objectIds }
-      .forEach { remove(it) }
+    pool.entries.removeIf { (key, _) -> key !in objectIds }
   }
 
   /**
@@ -104,7 +102,7 @@ internal class ObjectsPool(
   /**
    * Clears the data stored for all objects in the pool.
    */
-  fun clearObjectsData(emitUpdateEvents: Boolean) {
+  private fun clearObjectsData(emitUpdateEvents: Boolean) {
     for (obj in pool.values) {
       val update = obj.clearData()
       if (emitUpdateEvents) {
@@ -118,7 +116,7 @@ internal class ObjectsPool(
    *
    * @spec RTO6 - Creates zero-value objects when needed
    */
-  fun createZeroValueObjectIfNotExists(objectId: String): BaseLiveObject {
+  internal fun createZeroValueObjectIfNotExists(objectId: String): BaseLiveObject {
     val existingObject = get(objectId)
     if (existingObject != null) {
       return existingObject // RTO6a
@@ -148,23 +146,13 @@ internal class ObjectsPool(
    * Garbage collection interval handler.
    */
   private fun onGCInterval() {
-    val toDelete = mutableListOf<String>()
-
-    for ((objectId, obj) in pool.entries) {
-      // Tombstoned objects should be removed from the pool if they have been tombstoned for longer than grace period.
-      // By removing them from the local pool, Objects plugin no longer keeps a reference to those objects, allowing JVM's
-      // Garbage Collection to eventually free the memory for those objects, provided the user no longer references them either.
-      if (obj.isTombstoned &&
-          obj.tombstonedAt != null &&
-          System.currentTimeMillis() - obj.tombstonedAt!! >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS) {
-        toDelete.add(objectId)
-        continue
+    pool.entries.removeIf { (_, obj) ->
+      if (obj.isEligibleForGc()) { true } // Remove from pool
+      else {
+        obj.onGCInterval()
+        false  // Keep in pool
       }
-
-      obj.onGCInterval()
     }
-
-    toDelete.forEach { pool.remove(it) }
   }
 
   /**

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

@@ -1,34 +1,32 @@
 package io.ably.lib.objects.type
 
-import io.ably.lib.objects.LiveObjectsAdapter
+import io.ably.lib.objects.*
 import io.ably.lib.objects.ObjectMessage
 import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectState
-import io.ably.lib.objects.ObjectsPool
+import io.ably.lib.objects.ObjectsPoolDefaults
 import io.ably.lib.objects.objectError
 import io.ably.lib.util.Log
 
 /**
  * Base implementation of LiveObject interface.
  * Provides common functionality for all live objects.
  *
- * @spec RTLO1/RTLO2 - Base class for LiveMap/LiveCounter objects
+ * @spec RTLO1/RTLO2 - Base class for LiveMap/LiveCounter object
  */
 internal abstract class BaseLiveObject(
-  protected val objectId: String, // // RTLO3a
+  internal val objectId: String, // // RTLO3a
   protected val adapter: LiveObjectsAdapter
 ) {
 
   protected open val tag = "BaseLiveObject"
-  internal var isTombstoned = false
-  internal var tombstonedAt: Long? = null
 
-  protected val siteTimeserials = mutableMapOf<String, String>() // RTLO3b
+  internal val siteTimeserials = mutableMapOf<String, String>() // RTLO3b
 
-  /**
-   * @spec RTLO3 - Flag to track if create operation has been merged for LiveMap/LiveCounter
-   */
-  protected var createOperationIsMerged = false // RTLO3c
+  internal var createOperationIsMerged = false // RTLO3c
+
+  internal var isTombstoned = false
+  private var tombstonedAt: Long? = null
 
   fun notifyUpdated(update: Any) {
     // TODO: Implement event emission for updates
@@ -40,7 +38,7 @@ internal abstract class BaseLiveObject(
    *
    * @spec RTLO4a - Serial comparison logic for LiveMap/LiveCounter operations
    */
-  protected fun canApplyOperation(siteCode: String?, serial: String?): Boolean {
+  internal fun canApplyOperation(siteCode: String?, serial: String?): Boolean {
     if (serial.isNullOrEmpty()) {
       throw objectError("Invalid serial: $serial") // RTLO4a3
     }
@@ -51,41 +49,31 @@ internal abstract class BaseLiveObject(
     return existingSiteSerial == null || serial > existingSiteSerial // RTLO4a5, RTLO4a6
   }
 
-  /**
-   * Updates the time serial for a given site code.
-   */
-  protected fun updateTimeSerial(opSiteCode: String, opSerial: String) {
-    siteTimeserials[opSiteCode] = opSerial
-  }
-
-  /**
-   * Applies object delete operation.
-   *
-   * @spec RTLM10/RTLC10 - Object deletion for LiveMap/LiveCounter
-   */
-  protected fun applyObjectDelete(): Any {
-    return tombstone()
-  }
-
   /**
    * Marks the object as tombstoned.
-   *
-   * @spec RTLM11/RTLC11 - Tombstone functionality for LiveMap/LiveCounter
    */
-  protected fun tombstone(): Any {
+  internal fun tombstone(): Any {
     isTombstoned = true
     tombstonedAt = System.currentTimeMillis()
     val update = clearData()
     // TODO: Emit lifecycle events
     return update
   }
 
+  /**
+   * Checks if the object is eligible for garbage collection.
+   */
+  internal fun isEligibleForGc(): Boolean {
+    val currentTime = System.currentTimeMillis()
+    return isTombstoned && tombstonedAt?.let { currentTime - it >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS } == true
+  }
+
   /**
    * This is invoked by ObjectMessage having updated data with parent `ProtocolMessageAction` as `object_sync`
    * @return an update describing the changes
    * @spec RTLM6/RTLC6 - Overrides ObjectMessage with object data state from sync to LiveMap/LiveCounter
    */
-  abstract fun overrideWithObjectState(objectState: ObjectState): Map<String, Any>
+  abstract fun applyObjectState(objectState: ObjectState): Map<String, Any>
 
   /**
    * This is invoked by ObjectMessage having updated data with parent `ProtocolMessageAction` as `object`
@@ -96,11 +84,28 @@ internal abstract class BaseLiveObject(
 
   /**
    * Clears the object's data and returns an update describing the changes.
+   * This is called during tombstoning and explicit clear operations.
+   *
+   * This method:
+   * 1. Calculates a diff between the current state and an empty state
+   * 2. Clears all entries from the underlying data structure
+   * 3. Returns a map containing metadata about what was cleared
+   *
+   * The returned map is used to notifying other components about what entries were removed.
+   *
+   * @return A map representing the diff of changes made
    */
   abstract fun clearData(): Map<String, Any>
 
   /**
-   * Called during garbage collection intervals.
+   * Called during garbage collection intervals to clean up expired entries.
+   *
+   * This method should identify and remove entries that:
+   * - Have been marked as tombstoned
+   * - Have a tombstone timestamp older than the configured grace period
+   *
+   * Implementations typically use single-pass removal techniques to
+   * efficiently clean up expired data without creating temporary collections.
    */
   abstract fun onGCInterval()
 }

live-objects/src/main/kotlin/io/ably/lib/objects/type/livecounter/DefaultLiveCounter.kt

@@ -0,0 +1,82 @@
+package io.ably.lib.objects.type.livecounter
+
+import io.ably.lib.objects.*
+import io.ably.lib.objects.ObjectMessage
+import io.ably.lib.objects.ObjectOperation
+import io.ably.lib.objects.ObjectState
+import io.ably.lib.objects.type.BaseLiveObject
+import io.ably.lib.types.Callback
+
+/**
+ * Implementation of LiveObject for LiveCounter.
+ *
+ * @spec RTLC1/RTLC2 - LiveCounter implementation extends LiveObject
+ */
+internal class DefaultLiveCounter(
+  objectId: String,
+  adapter: LiveObjectsAdapter,
+) : LiveCounter, BaseLiveObject(objectId, adapter) {
+
+  override val tag = "LiveCounter"
+
+  /**
+   * Counter data value
+   */
+  internal var data: Long = 0 // RTLC3
+
+  /**
+   * liveCounterManager instance for managing LiveMap operations
+   */
+  private val liveCounterManager = LiveCounterManager(this)
+
+  override fun increment() {
+    TODO("Not yet implemented")
+  }
+
+  override fun incrementAsync(callback: Callback<Void>) {
+    TODO("Not yet implemented")
+  }
+
+  override fun decrement() {
+    TODO("Not yet implemented")
+  }
+
+  override fun decrementAsync(callback: Callback<Void>) {
+    TODO("Not yet implemented")
+  }
+
+  /**
+   * @spec RTLC5 - Returns the current counter value
+   */
+  override fun value(): Long {
+    // RTLC5a, RTLC5b - Configuration validation would be done here
+    return data // RTLC5c
+  }
+
+  override fun applyObjectState(objectState: ObjectState): Map<String, Long> {
+    return liveCounterManager.applyObjectState(objectState)
+  }
+
+  override fun applyOperation(operation: ObjectOperation, message: ObjectMessage) {
+    liveCounterManager.applyOperation(operation, message)
+  }
+
+  override fun clearData(): Map<String, Long> {
+    return mapOf("amount" to data).apply { data = 0 }
+  }
+
+  override fun onGCInterval() {
+    // Nothing to GC for a counter object
+    return
+  }
+
+  companion object {
+    /**
+     * Creates a zero-value counter object.
+     * @spec RTLC4 - Returns LiveCounter with 0 value
+     */
+    internal fun zeroValue(objectId: String, adapter: LiveObjectsAdapter): DefaultLiveCounter {
+      return DefaultLiveCounter(objectId, adapter)
+    }
+  }
+}