[ECO-5426] Added separate LivemapManager 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

@@ -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

@@ -104,7 +104,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 +118,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 +148,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,10 +1,10 @@
 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
 
@@ -15,20 +15,18 @@ import io.ably.lib.util.Log
  * @spec RTLO1/RTLO2 - Base class for LiveMap/LiveCounter objects
  */
 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
+  internal 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/DefaultLiveCounter.kt

@@ -33,7 +33,7 @@ internal class DefaultLiveCounter(
   /**
    * @spec RTLC6 - Overrides counter data with state from sync
    */
-  override fun overrideWithObjectState(objectState: ObjectState): Map<String, Long> {
+  override fun applyObjectState(objectState: ObjectState): Map<String, Long> {
     if (objectState.objectId != objectId) {
       throw objectError("Invalid object state: object state objectId=${objectState.objectId}; LiveCounter objectId=$objectId")
     }
@@ -94,7 +94,7 @@ internal class DefaultLiveCounter(
     }
     // should update stored site serial immediately. doesn't matter if we successfully apply the op,
     // as it's important to mark that the op was processed by the object
-    updateTimeSerial(opSiteCode!!, opSerial!!) // RTLC7c
+    siteTimeserials[opSiteCode!!] = opSerial!! // RTLC7c
 
     if (isTombstoned) {
       // this object is tombstoned so the operation cannot be applied
@@ -110,7 +110,7 @@ internal class DefaultLiveCounter(
           throw payloadError(operation)
         }
       }
-      ObjectOperationAction.ObjectDelete -> applyObjectDelete()
+      ObjectOperationAction.ObjectDelete -> tombstone()
       else -> throw objectError("Invalid ${operation.action} op for LiveCounter objectId=${objectId}") // RTLC7d3
     }
 
@@ -165,10 +165,6 @@ internal class DefaultLiveCounter(
     return mapOf("amount" to count)
   }
 
-  /**
-   * Called during garbage collection intervals.
-   * Nothing to GC for a counter object.
-   */
   override fun onGCInterval() {
     // Nothing to GC for a counter object
     return